shuim
a264561576
- 스윙 체결가 확인 전 executed_price NULL + fill_sweep, 진입가 resolve(ActiveTrade↔Portfolio), 0체결/부분체결 매수 장부 원복 - DT 고아 감지(근거 한정)·0체결 매도 복원·stale 점검 확대, 시간손절 -0.3% 원복 - 폴백 손절 ATR 캐시, 거래량부족-차단 신호 매수 차단, 현금 5% 유보 재적용, MDD peak/reset 수정 - KIS 서킷브레이커 청산 우회·경보, db_worker 재큐잉 순서, heartbeat·health-watch 사이드카 - 실시간 WS 재구독·41건 우선순위·다건 프레임 분할, 차트 샘플 제거·당일봉 합성, WS 서브프로토콜 인증 - 프론트: 미관리 그룹·확인 모달·차트 실시간 반영·예수금 표시 정정 - 전문가 에이전트 6종 추가(.claude/agents) - 유닛 스위트 2667 passed (소켓 차단 하), tsc OK Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Y83Za98XsNw9zuA4FTUytN
508 lines
23 KiB
Python
508 lines
23 KiB
Python
"""
|
|
연속 손실 추적 - CLAUDE.md 섹션 7.4
|
|
|
|
동일 종목에서 3회 이상 연속 손실 시 30거래일 동안 신규 진입 금지.
|
|
재시작 후에도 ban 상태 유지: DB(stock_bans 테이블) 영속화.
|
|
"""
|
|
import asyncio
|
|
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Dict, Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 기동 복원 재시도 간격(초) — 테스트에서 0 으로 덮어쓴다.
|
|
_LOAD_RETRY_DELAYS = (1.0, 2.0)
|
|
# 복원 실패 후 진입 경로에서의 재시도 최소 간격(초)
|
|
_LOAD_RETRY_MIN_INTERVAL_SEC = 30.0
|
|
# 백그라운드 재복원 1회의 전체 상한(초) — 느린 DB 에서 태스크가 무한정 매달리지 않게 한다.
|
|
_LOAD_RETRY_TASK_TIMEOUT_SEC = 15.0
|
|
|
|
# tz-aware/naive 경계 정규화 — app/core/dt_utils 참고 (db_worker.py와 공유)
|
|
from app.core.dt_utils import to_naive_utc as _to_naive_utc, to_aware_utc as _to_aware_utc # noqa: E402
|
|
|
|
|
|
class ConsecutiveLossTracker:
|
|
"""종목별 연속 손실 추적 및 쿨링오프 관리"""
|
|
|
|
# 멀티테넌트: risk_factory.create_loss_tracker_for_user()가 인스턴스 생성 후
|
|
# 세팅하는 소유자 user_id. 레거시 싱글톤은 None으로 유지(전역 stock_bans 사용).
|
|
user_id: Optional[int] = None
|
|
|
|
def __init__(self, ban_count: int = 3, ban_calendar_days: int = 42):
|
|
"""
|
|
Args:
|
|
ban_count: 연속 손실 횟수 임계값 (기본 3)
|
|
ban_calendar_days: 금지 기간 (30거래일 ≈ 42캘린더일)
|
|
"""
|
|
self.ban_count = ban_count
|
|
self.ban_calendar_days = ban_calendar_days
|
|
# {stock_code: {"consecutive_losses": int, "ban_until": datetime|None}}
|
|
self._tracker: Dict[str, dict] = {}
|
|
self._lock = asyncio.Lock() # [FIX] 동시 접근 방지용 비동기 락
|
|
# DB 복원 실패 상태(fail-closed): True 인 동안 is_banned() 는 모든 종목을 금지로 본다.
|
|
# 기본 False — load_from_db 를 호출하지 않는 인스턴스(테스트·MT)는 영향 없음.
|
|
self._load_failed: bool = False
|
|
self._last_load_attempt: float = 0.0
|
|
# 백그라운드 재복원 태스크(단일 비행)
|
|
self._reload_task: Optional["asyncio.Task"] = None
|
|
# 복원 실패 중 기록되어 DB 저장이 보류된 종목(복원 성공 후 병합·저장)
|
|
self._dirty: set = set()
|
|
# 복원 실패 중 종목별 결과 이벤트(순서 보존, True=손실). 복원 성공 시 DB 상태 위에 재생한다.
|
|
self._pending_events: Dict[str, list] = {}
|
|
|
|
def record_trade_result(self, stock_code: str, is_loss: bool):
|
|
"""매매 결과 인메모리 기록 (DB 저장은 save_to_db 별도 호출)
|
|
|
|
시그니처 불변. 구현은 record_trade_result_v2 로 위임한다(유효한 ban 보존).
|
|
|
|
Args:
|
|
stock_code: 종목코드
|
|
is_loss: 손실 여부
|
|
"""
|
|
self.record_trade_result_v2(stock_code, is_loss)
|
|
|
|
def record_trade_result_v2(self, stock_code: str, is_loss: bool):
|
|
"""[v2] 매매 결과 기록 — 유효한 ban 은 이익 청산으로 해제되지 않는다.
|
|
|
|
ban_until > now 인 동안 이익 기록은 ban_until/연패 카운터를 모두 유지한다
|
|
(밴 이전 포지션·외부 유입 보유분의 이익 청산이 42일 금지를 풀던 결함).
|
|
ban 만료 후의 이익 기록만 카운터를 리셋한다.
|
|
"""
|
|
if stock_code not in self._tracker:
|
|
self._tracker[stock_code] = {"consecutive_losses": 0, "ban_until": None}
|
|
|
|
if self._load_failed:
|
|
# DB 기준선 없이 0 부터 센 증분이므로 이벤트 순서를 보존해 복원 후 재생한다.
|
|
self._pending_events.setdefault(stock_code, []).append(bool(is_loss))
|
|
self._dirty.add(stock_code)
|
|
|
|
self._apply_result_impl(stock_code, self._tracker[stock_code], is_loss)
|
|
|
|
def _apply_result_impl(self, stock_code: str, entry: dict, is_loss: bool):
|
|
"""단일 결과를 entry 에 적용(record_trade_result_v2 본체, 복원 후 재생에도 사용)."""
|
|
if is_loss:
|
|
_bu = entry.get("ban_until")
|
|
_now = datetime.now(timezone.utc)
|
|
_ban_active = _bu is not None and _now < _to_aware_utc(_bu)
|
|
if _bu is not None and not _ban_active:
|
|
# 만료된 밴: 카운터를 먼저 리셋(손실 1건으로 즉시 재밴 방지)
|
|
entry["consecutive_losses"] = 0
|
|
entry["ban_until"] = None
|
|
entry["consecutive_losses"] += 1
|
|
if _ban_active:
|
|
# 유효 밴 중 추가 손실은 ban_until 을 연장하지 않는다
|
|
return
|
|
if entry["consecutive_losses"] >= self.ban_count:
|
|
entry["ban_until"] = _now + timedelta(days=self.ban_calendar_days)
|
|
logger.warning(
|
|
f"{stock_code}: {entry['consecutive_losses']}회 연속 손실 → "
|
|
f"{entry['ban_until'].strftime('%Y-%m-%d')}까지 진입 금지"
|
|
)
|
|
else:
|
|
ban_until = entry.get("ban_until")
|
|
if ban_until is not None and datetime.now(timezone.utc) < _to_aware_utc(ban_until):
|
|
logger.info(
|
|
f"{stock_code}: 진입 금지 유효({ban_until.strftime('%Y-%m-%d')}까지) — "
|
|
f"이익 청산이지만 ban 유지"
|
|
)
|
|
return
|
|
entry["consecutive_losses"] = 0
|
|
entry["ban_until"] = None
|
|
|
|
async def save_to_db(self, db=None, stock_code: str = None):
|
|
"""인메모리 상태를 DB에 영속화 (record_trade_result 후 호출)
|
|
|
|
NOTE: db 파라미터는 하위호환성을 위해 유지되지만 사용되지 않습니다.
|
|
실제 DB 저장은 db_worker.py의 db_task_queue를 통해 처리됩니다.
|
|
"""
|
|
if stock_code is None:
|
|
return
|
|
|
|
try:
|
|
from app.services.db_worker import enqueue_stock_ban_upsert
|
|
|
|
entry = self._tracker.get(stock_code)
|
|
if entry is None:
|
|
return
|
|
|
|
if self._load_failed:
|
|
# 복원 실패 상태에서 저장하면 DB 의 기존 밴 행을 덮어써 42일 밴이 소실된다.
|
|
self._dirty.add(stock_code)
|
|
logger.error(
|
|
f"loss_tracker 복원 실패 상태 — {stock_code} DB 저장 보류(복원 성공 후 병합 저장)"
|
|
)
|
|
return
|
|
|
|
# 큐에 UPSERT 작업 등록 (db_worker가 비동기적으로 처리)
|
|
# ban_until 컬럼은 tz-naive이므로 저장 직전 naive-UTC로 정규화한다.
|
|
await enqueue_stock_ban_upsert(
|
|
stock_code=stock_code,
|
|
consecutive_losses=entry["consecutive_losses"],
|
|
ban_until=_to_naive_utc(entry["ban_until"]),
|
|
)
|
|
logger.debug(f"loss_tracker DB 저장 큐 등록: {stock_code}")
|
|
except Exception as e:
|
|
logger.error(f"loss_tracker DB 저장 큐 등록 실패 ({stock_code}): {e}")
|
|
|
|
async def save_to_db_v2(self, user_id: int, stock_code: str = None):
|
|
"""[v2] 사용자 스코프 인메모리 상태를 stock_bans_v2 에 영속화.
|
|
|
|
save_to_db()와 동일 패턴이지만 db_worker의 upsert_stock_ban_v2 task로
|
|
(user_id, stock_code) 단위 UPSERT한다. 기존 save_to_db()는 불변.
|
|
"""
|
|
if stock_code is None:
|
|
return
|
|
|
|
try:
|
|
from app.services.db_worker import enqueue_stock_ban_upsert_v2
|
|
|
|
entry = self._tracker.get(stock_code)
|
|
if entry is None:
|
|
return
|
|
|
|
# stock_bans_v2.ban_until 도 tz-naive 컬럼 — 저장 직전 naive-UTC로 정규화.
|
|
await enqueue_stock_ban_upsert_v2(
|
|
user_id=user_id,
|
|
stock_code=stock_code,
|
|
consecutive_losses=entry["consecutive_losses"],
|
|
ban_until=_to_naive_utc(entry["ban_until"]),
|
|
)
|
|
logger.debug(f"loss_tracker_v2 DB 저장 큐 등록: user={user_id} {stock_code}")
|
|
except Exception as e:
|
|
logger.error(f"loss_tracker_v2 DB 저장 큐 등록 실패 (user={user_id}, {stock_code}): {e}")
|
|
|
|
async def load_from_db_v2(self, db, user_id: int):
|
|
"""[v2] 앱 시작/컨텍스트 생성 시 stock_bans_v2 에서 사용자 스코프 ban 상태 복원."""
|
|
try:
|
|
from sqlalchemy import select
|
|
from app.models.models import StockBanV2
|
|
|
|
result = await db.execute(select(StockBanV2).where(StockBanV2.user_id == user_id))
|
|
records = result.scalars().all()
|
|
|
|
now = datetime.now(timezone.utc)
|
|
loaded = 0
|
|
for rec in records:
|
|
# DB의 ban_until은 tz-naive → tz-aware now와 비교하기 전 정규화.
|
|
ban_until = _to_aware_utc(rec.ban_until)
|
|
if ban_until and now >= ban_until:
|
|
ban_until = None
|
|
consecutive_losses = 0
|
|
else:
|
|
consecutive_losses = rec.consecutive_losses
|
|
|
|
self._tracker[rec.stock_code] = {
|
|
"consecutive_losses": consecutive_losses,
|
|
"ban_until": ban_until,
|
|
}
|
|
if ban_until:
|
|
loaded += 1
|
|
logger.info(
|
|
f"ban_v2 복원: user={user_id} {rec.stock_code} → "
|
|
f"{ban_until.strftime('%Y-%m-%d')}까지 진입 금지"
|
|
)
|
|
|
|
logger.info(
|
|
f"loss_tracker_v2 DB 로드 완료: user={user_id} {len(records)}건 중 {loaded}건 활성 ban"
|
|
)
|
|
except Exception:
|
|
logger.exception(f"loss_tracker_v2 DB 로드 실패 (user={user_id})")
|
|
|
|
async def load_from_db(self, db):
|
|
"""앱 시작 시 DB에서 ban 상태 복원 (시그니처 불변).
|
|
|
|
실패 시 짧게 재시도하고, 끝내 실패하면 _load_failed=True (fail-closed: 신규 진입 차단)
|
|
+ 텔레그램 경고. 성공하면 플래그 해제.
|
|
"""
|
|
import time as _time
|
|
|
|
last_err = None
|
|
attempts = len(_LOAD_RETRY_DELAYS) + 1
|
|
for attempt in range(attempts):
|
|
try:
|
|
await self._load_from_db_once(db)
|
|
self._last_load_attempt = _time.monotonic()
|
|
if self._load_failed:
|
|
logger.warning("loss_tracker DB 복원 성공 — fail-closed 해제")
|
|
self._load_failed = False
|
|
self._pending_events.clear()
|
|
await self._flush_dirty()
|
|
return
|
|
except Exception as e:
|
|
last_err = e
|
|
logger.warning(f"loss_tracker DB 로드 실패 ({attempt + 1}/{attempts}): {e}")
|
|
try:
|
|
await db.rollback()
|
|
except Exception as rb:
|
|
logger.warning(f"loss_tracker 로드 실패 후 rollback 실패: {rb}")
|
|
if attempt < attempts - 1:
|
|
await asyncio.sleep(_LOAD_RETRY_DELAYS[attempt])
|
|
|
|
self._last_load_attempt = _time.monotonic()
|
|
first = not self._load_failed
|
|
self._load_failed = True
|
|
logger.error(
|
|
f"loss_tracker DB 로드 최종 실패 — 신규 진입 차단(fail-closed), 저장 보류: {last_err}"
|
|
)
|
|
if first:
|
|
try:
|
|
from app.services.telegram_notifier import telegram_notifier
|
|
await telegram_notifier.notify_system_warning(
|
|
"LOSS_TRACKER",
|
|
"연속손실 밴 DB 복원 실패 - 복원될 때까지 신규 진입 차단",
|
|
)
|
|
except Exception as ne:
|
|
logger.warning(f"loss_tracker 복원 실패 텔레그램 알림 실패: {ne}")
|
|
|
|
async def _load_from_db_once(self, db):
|
|
"""단일 시도 — 실패 시 예외를 그대로 올린다."""
|
|
from sqlalchemy import select
|
|
from app.models.models import StockBan
|
|
|
|
result = await db.execute(select(StockBan))
|
|
records = result.scalars().all()
|
|
|
|
now = datetime.now(timezone.utc)
|
|
loaded = 0
|
|
for rec in records:
|
|
# 만료된 ban은 초기화. DB의 ban_until은 tz-naive → 비교 전 정규화.
|
|
ban_until = _to_aware_utc(rec.ban_until)
|
|
if ban_until and now >= ban_until:
|
|
ban_until = None
|
|
consecutive_losses = 0
|
|
else:
|
|
consecutive_losses = rec.consecutive_losses
|
|
|
|
new_entry = {
|
|
"consecutive_losses": consecutive_losses,
|
|
"ban_until": ban_until,
|
|
}
|
|
events = self._pending_events.get(rec.stock_code)
|
|
if events:
|
|
# 복원 실패 중 기록된 결과를 DB 상태 위에 순서대로 재생(밴 판정 포함)
|
|
for _is_loss in events:
|
|
self._apply_result_impl(rec.stock_code, new_entry, _is_loss)
|
|
ban_until = new_entry["ban_until"]
|
|
self._tracker[rec.stock_code] = new_entry
|
|
if ban_until:
|
|
loaded += 1
|
|
logger.info(
|
|
f"ban 복원: {rec.stock_code} → {ban_until.strftime('%Y-%m-%d')}까지 진입 금지"
|
|
)
|
|
|
|
logger.info(f"loss_tracker DB 로드 완료: {len(records)}건 중 {loaded}건 활성 ban")
|
|
|
|
async def _flush_dirty(self):
|
|
"""복원 성공 후 보류됐던 저장을 큐에 등록."""
|
|
dirty, self._dirty = self._dirty, set()
|
|
for code in dirty:
|
|
await self.save_to_db(None, code)
|
|
|
|
async def retry_load_if_failed_v2(self) -> bool:
|
|
"""[v2] 복원 실패 상태면(최소 간격 경과 시) 백그라운드 재복원을 예약한다. 복원 정상이면 True.
|
|
|
|
실시간 진입 경로에서 호출되므로 DB 를 기다리지 않고 즉시 반환한다.
|
|
재복원은 단일 비행 백그라운드 태스크(SELECT 전용)가 수행하고 결과는
|
|
_load_failed 플래그로만 반영된다. 진입 판정은 플래그만 보고 즉시 fail-closed 한다.
|
|
"""
|
|
if not self._load_failed:
|
|
return True
|
|
import time as _time
|
|
task = self._reload_task
|
|
if task is not None and not task.done():
|
|
return False # 이미 진행 중
|
|
if _time.monotonic() - self._last_load_attempt < _LOAD_RETRY_MIN_INTERVAL_SEC:
|
|
return False
|
|
self._last_load_attempt = _time.monotonic()
|
|
try:
|
|
self._reload_task = asyncio.get_running_loop().create_task(self._reload_in_background_impl())
|
|
except RuntimeError as e:
|
|
logger.warning(f"loss_tracker 재복원 태스크 생성 실패: {e}")
|
|
return False
|
|
|
|
async def _reload_in_background_impl(self) -> None:
|
|
"""백그라운드 재복원 본체 — wait_for 상한 적용, 예외는 warning."""
|
|
try:
|
|
# DB-READ-EXCEPTION: stock_bans SELECT 전용 복원 재시도(commit 없음), 진입 경로 밖 백그라운드
|
|
from app.core.database import AsyncSessionLocal
|
|
|
|
async def _run():
|
|
async with AsyncSessionLocal() as session:
|
|
await self.load_from_db(session)
|
|
|
|
await asyncio.wait_for(_run(), timeout=_LOAD_RETRY_TASK_TIMEOUT_SEC)
|
|
except asyncio.TimeoutError:
|
|
logger.warning(
|
|
f"loss_tracker 복원 재시도 시간초과({_LOAD_RETRY_TASK_TIMEOUT_SEC}s) — fail-closed 유지"
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"loss_tracker 복원 재시도 실패: {e}")
|
|
|
|
def is_banned(self, stock_code: str) -> bool:
|
|
"""해당 종목이 진입 금지 상태인지 확인 (DB 복원 실패 중에는 fail-closed 로 True)"""
|
|
if self._load_failed:
|
|
return True
|
|
entry = self._tracker.get(stock_code)
|
|
if not entry or not entry["ban_until"]:
|
|
return False
|
|
if datetime.now(timezone.utc) >= entry["ban_until"]:
|
|
# 금지 기간 만료 → 리셋
|
|
entry["consecutive_losses"] = 0
|
|
entry["ban_until"] = None
|
|
return False
|
|
return True
|
|
|
|
async def is_banned_safe(self, stock_code: str) -> bool:
|
|
"""[v2] 비동기 락 보호 is_banned — 동시 접근이 우려되는 경로에서 사용"""
|
|
if self._load_failed:
|
|
await self.retry_load_if_failed_v2() # 락 밖, SELECT 전용
|
|
async with self._lock:
|
|
return self.is_banned(stock_code)
|
|
|
|
async def record_trade_result_safe(self, stock_code: str, is_loss: bool):
|
|
"""[v2] 비동기 락 보호 record_trade_result"""
|
|
async with self._lock:
|
|
self.record_trade_result_v2(stock_code, is_loss)
|
|
|
|
async def get_all_bans_safe(self) -> list:
|
|
"""[v2] Lock 보호 하에 현재 활성 ban 종목 코드 목록 반환.
|
|
|
|
Returns:
|
|
ban_until이 아직 유효한 종목코드 리스트
|
|
"""
|
|
async with self._lock:
|
|
now = datetime.now(timezone.utc)
|
|
return [
|
|
code for code, entry in self._tracker.items()
|
|
if entry.get("ban_until") and now < entry["ban_until"]
|
|
]
|
|
|
|
def get_ban_info(self, stock_code: str) -> dict:
|
|
"""종목의 연속 손실 및 금지 정보 조회"""
|
|
entry = self._tracker.get(stock_code, {"consecutive_losses": 0, "ban_until": None})
|
|
return {
|
|
"stock_code": stock_code,
|
|
"consecutive_losses": entry["consecutive_losses"],
|
|
"is_banned": self.is_banned(stock_code),
|
|
"ban_until": entry["ban_until"].isoformat() if entry["ban_until"] else None,
|
|
}
|
|
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# 1회성 소급 보정 (2026-10-10, 라운드 7)
|
|
#
|
|
# 10/01 시간손절 중립화는 기록 시점에만 적용돼, 그 이전에 손실로 센 시간손절이
|
|
# stock_bans.consecutive_losses 에 남아 있다. 아래 함수들은 trade_history 를 현행 규칙
|
|
# (시간손절 제외, 비용 차감 순손익)으로 재생해 "현재 DB 행 vs 재계산 결과" 차이표를 만든다.
|
|
# 자동 일괄 해제는 하지 않는다: plan 생성은 순수 계산(DB/네트워크 없음)이고, 적용은
|
|
# 운영자가 승인한 종목코드만 db_task_queue(upsert_stock_ban) 경유로 쓴다.
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
|
|
def replay_streak_v2(events, ban_count: int, ban_calendar_days: int) -> dict:
|
|
"""청산 이벤트를 시간순으로 재생해 현행 규칙의 연패/밴 상태를 계산(순수 함수).
|
|
|
|
events: 시간순(오름차순)이어야 하는 dict 리스트. 키
|
|
- at: datetime (체결/청산 시각)
|
|
- net: 왕복 순손익(float, 분할 익절 레그는 호출자가 합산해 전량 청산 단위로 전달)
|
|
- time_stop: bool (True 면 중립 — 건너뜀)
|
|
규칙은 ConsecutiveLossTracker._apply_result_impl 과 동일(유효 밴 중 이익은 밴 유지,
|
|
유효 밴 중 손실은 ban_until 비연장, 만료 후 손실은 카운터 리셋 뒤 1부터).
|
|
반환: {"consecutive_losses": int, "ban_until": datetime|None}
|
|
"""
|
|
cnt = 0
|
|
ban_until = None
|
|
for ev in events:
|
|
if ev.get("time_stop"):
|
|
continue
|
|
at = ev["at"]
|
|
is_loss = float(ev["net"]) < 0
|
|
active = ban_until is not None and at < ban_until
|
|
if is_loss:
|
|
if ban_until is not None and not active:
|
|
cnt = 0
|
|
ban_until = None
|
|
cnt += 1
|
|
if active:
|
|
continue
|
|
if cnt >= ban_count:
|
|
ban_until = at + timedelta(days=ban_calendar_days)
|
|
else:
|
|
if active:
|
|
continue
|
|
cnt = 0
|
|
ban_until = None
|
|
return {"consecutive_losses": cnt, "ban_until": ban_until}
|
|
|
|
|
|
def build_ban_correction_plan_v2(events_by_code: dict, current_rows: dict,
|
|
ban_count: int, ban_calendar_days: int) -> list:
|
|
"""종목별 현재 stock_bans 행 vs 현행 규칙 재계산 결과 차이표(쓰기 없음).
|
|
|
|
current_rows: {stock_code: {"consecutive_losses": int, "ban_until": datetime|None}}
|
|
반환: 차이가 있는 종목만 [{stock_code, current, recomputed, ban_lifted}] (코드순).
|
|
ban_lifted=True 는 현재 밴이 재계산에서는 없는 경우 — 운영자가 개별 승인할 대상.
|
|
"""
|
|
plan = []
|
|
for code in sorted(set(events_by_code) | set(current_rows)):
|
|
cur = current_rows.get(code) or {"consecutive_losses": 0, "ban_until": None}
|
|
new = replay_streak_v2(events_by_code.get(code, []), ban_count, ban_calendar_days)
|
|
if (cur.get("consecutive_losses", 0) == new["consecutive_losses"]
|
|
and cur.get("ban_until") == new["ban_until"]):
|
|
continue
|
|
plan.append({
|
|
"stock_code": code,
|
|
"current": dict(cur),
|
|
"recomputed": new,
|
|
"ban_lifted": cur.get("ban_until") is not None and new["ban_until"] is None,
|
|
})
|
|
return plan
|
|
|
|
|
|
async def apply_ban_correction_v2(plan: list, approved_codes) -> list:
|
|
"""운영자가 승인한 종목만 db_task_queue(upsert_stock_ban) 로 보정 기록. 적용 종목 반환.
|
|
|
|
approved_codes 가 비어 있으면 아무것도 쓰지 않는다. plan 에 없는 코드는 무시한다.
|
|
인메모리 loss_tracker 는 건드리지 않는다(적용 후 재기동/load_from_db 로 반영).
|
|
"""
|
|
approved = {str(c) for c in (approved_codes or [])}
|
|
if not approved:
|
|
return []
|
|
from app.services.db_worker import enqueue_stock_ban_upsert
|
|
|
|
applied = []
|
|
for row in plan:
|
|
code = row["stock_code"]
|
|
if code not in approved:
|
|
continue
|
|
rec = row["recomputed"]
|
|
await enqueue_stock_ban_upsert(
|
|
stock_code=code,
|
|
consecutive_losses=rec["consecutive_losses"],
|
|
ban_until=_to_naive_utc(rec["ban_until"]),
|
|
)
|
|
applied.append(code)
|
|
return applied
|
|
|
|
|
|
# 싱글톤 인스턴스 — settings에서 ban_count 주입
|
|
def _create_loss_tracker() -> ConsecutiveLossTracker:
|
|
try:
|
|
from app.core.config import settings
|
|
# [HIGH-3 수정] 하드코딩 42 → settings.CONSECUTIVE_LOSS_BAN_DAYS(기본 42캘린더일).
|
|
# 운영자가 config/환경변수로 ban 기간을 조정해도 실제 동작에 반영되지 않던 문제 해소.
|
|
# 설정값이 비정상(<=0)이면 문서 기준값 42로 폴백한다.
|
|
_ban_days = int(getattr(settings, "CONSECUTIVE_LOSS_BAN_DAYS", 42) or 42)
|
|
if _ban_days <= 0:
|
|
_ban_days = 42
|
|
return ConsecutiveLossTracker(
|
|
ban_count=settings.CONSECUTIVE_LOSS_BAN_COUNT,
|
|
ban_calendar_days=_ban_days, # 30거래일 ≈ 42캘린더일 (CLAUDE.md §5.1)
|
|
)
|
|
except Exception:
|
|
return ConsecutiveLossTracker()
|
|
|
|
|
|
loss_tracker = _create_loss_tracker()
|