fix: 재기동 내구성 + DT 진입 복구 + 유입 잔고 격리 + 대시보드 표시 오류
## 재기동 내구성 - scheduler: 복원 직후 reset_if_new_day()가 복원분을 전부 폐기하던 버그 (begin_restore()로 날짜 경계를 선확정. 복원 성공 로그는 뜨는데 실제로는 지워졌다) - 일일 안전장치(daily_pnl/loss_streak_today/closed_today) Redis 영속화 날짜를 키와 페이로드 양쪽에 실어 이중 게이트, 복원은 보수적 병합 (손익 min/연패 max/청산목록 합집합 — 복원이 안전장치를 약화시킬 수 없다) - db_worker 종료 drain이 insert_trade_history만 커밋하고 포지션 추적 행은 DLQ로 버리던 비대칭 제거 + DLQ 재처리 경로 신설(쓰기 전용이었다) - 장중 기동 시 ORB 소급 수집(backfill), 전일 캔들 혼입 차단 - swing/exit_service "보유 종목 없음" debug→info, 10회 연속 warning ## DT 진입 복구 - 워밍업 회귀 복구: 1b9257c(2026-09-28)가 조회 계층에 당일 필터를 넣어 5분봉 RSI(14)/BB(20) 워밍업이 죽고 VWAP/RSI_BB가 10:15/10:40 이전 발화 불가가 됐다. 필터를 삭제하지 않고 용도별로 분리 (get_minute_chart_full_day_v2 신설, 기존 메서드·3개 호출부 동작 무변경) - 거래량 게이트 비교 대상 교정: 형성 중 봉 → 완성봉 (5분봉·ORB 1분봉 양쪽) 임계값(1.3/1.5)은 불변, 고친 것은 비교 대상뿐 - 세션 경계 차단: _last_session_candles()로 거래량 비교가 전일 봉을 잡아 개장 직후 게이트가 자동 통과하는 것을 방지 - 스캔 선조회(동시성 4)·종목 타임아웃·시간예산·라운드로빈 + 잘린 종목 수 로그 - watchlist 30분 갱신 시 ORB 보유 종목 우선 유지 ## 유입 잔고 격리 - sync_portfolio()가 시스템 매수 근거 없는 KIS 잔고를 자기 스윙 포지션으로 편입해 자동 손절로 처분하던 문제 → strategy_type='external' 격리 (승격만 허용, 강등 없음. 기존 행은 수동 SQL 필요) - 격리분 손절선 이탈은 집행 없이 warning + summary 노출 - 수동 청산 경로에 record_trade_result 누락 보강, 전략 간 당일 쿨다운 ## 4% 상한 - market_data 표시 경로 클램프 누락 + confidence 배율 후 재클램프 - 회귀 방지 테스트(AST로 가드가 try 밖에 있는지 검증) ## 프론트 - 총자산 보정 가드(3%·종목수·가격결손·국내분 한정), 원인 미규명 명시 - cumulative_realized_pl이 /portfolio/combined에 없어 항상 0이던 문제 테스트 579 → 677 passed / 0 failed (소켓 차단 하 통과, 실호출 없음) 전략 임계값 변경 0건 ⚠️ 워밍업 복구의 효과 크기는 미확정 — KIS 분봉 API에 날짜 파라미터가 없어 미래 ref_time에 전일 데이터를 주는지가 미검증 전제다. 배포 후 "[분봉] 워밍업 5분봉 N봉 (당일 a + 전일 b)" 로그의 b로 확정할 것. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -30,6 +30,7 @@ Supported task types:
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
import uuid
|
||||
from datetime import date, datetime
|
||||
from pathlib import Path
|
||||
@@ -116,6 +117,163 @@ def _apply_strategy_name_filter(query, data: dict, column):
|
||||
DLQ_DIR = Path.home() / ".cache" / "auto-stock" / "dlq"
|
||||
|
||||
|
||||
# 종료 drain 시간 예산(초). 초과분은 커밋을 포기하고 DLQ 로 보관한다.
|
||||
DRAIN_DEADLINE_SECONDS = 20
|
||||
# DLQ 재처리 완료 파일 이동 위치 — 재처리 무한루프 방지
|
||||
DLQ_PROCESSED_DIR = DLQ_DIR / "processed"
|
||||
# 한 번의 기동에서 재큐잉할 최대 DLQ 파일 수 (큐 폭주 방지)
|
||||
DLQ_REPLAY_MAX_FILES = 500
|
||||
|
||||
|
||||
# [M5 2026-09-28] DLQ 재처리 시 타입 복원.
|
||||
# _save_to_dlq 는 json.dumps(..., default=str) 를 쓰므로 Decimal → "1234.5",
|
||||
# datetime/date → ISO 문자열로 **열화**된다. 그 dict 를 그대로 큐에 되돌리면
|
||||
# `TradeHistory(**data)` 같은 ORM 대입에서 asyncpg 가 DateTime 컬럼의 str 을 거부해
|
||||
# 3회 재시도 후 **새 DLQ 파일로 다시 쌓인다**(기동마다 증식). 재큐잉 전에 모델 컬럼
|
||||
# 타입을 보고 되돌린다.
|
||||
_DLQ_TASK_MODELS: dict[str, str] = {
|
||||
"insert_trade_history": "TradeHistory",
|
||||
"update_trade_history": "TradeHistory",
|
||||
"update_trade_history_by_order_id": "TradeHistory",
|
||||
"upsert_portfolio": "Portfolio",
|
||||
"upsert_active_trade": "ActiveTrade",
|
||||
"close_active_trade": "ActiveTrade",
|
||||
"upsert_stock_analysis": "StockAnalysis",
|
||||
"bulk_upsert_stock_analysis": "StockAnalysis",
|
||||
"insert_trade_journal": "TradeJournal",
|
||||
"upsert_stock_ban": "StockBan",
|
||||
"upsert_stock_ban_v2": "StockBanV2",
|
||||
"insert_news": "News",
|
||||
}
|
||||
|
||||
|
||||
def _coerce_value_for_column(value, col_type):
|
||||
"""컬럼 타입에 맞춰 문자열을 원래 타입으로 되돌린다. 실패 시 원본 그대로."""
|
||||
if not isinstance(value, str) or not value:
|
||||
return value
|
||||
try:
|
||||
from sqlalchemy import DateTime, Date, Numeric, Float, Integer, Boolean
|
||||
if isinstance(col_type, DateTime):
|
||||
return datetime.fromisoformat(value)
|
||||
if isinstance(col_type, Date):
|
||||
return date.fromisoformat(value)
|
||||
if isinstance(col_type, Numeric) and not isinstance(col_type, Float):
|
||||
from decimal import Decimal
|
||||
return Decimal(value)
|
||||
if isinstance(col_type, Float):
|
||||
return float(value)
|
||||
if isinstance(col_type, Integer):
|
||||
return int(value)
|
||||
if isinstance(col_type, Boolean):
|
||||
return value.lower() in ("true", "1", "yes")
|
||||
except (ValueError, TypeError, ArithmeticError):
|
||||
# 되돌리지 못하면 원본을 유지 — 핸들러가 판단하게 한다(조용히 버리지 않는다)
|
||||
logger.warning("[DLQ] 타입 복원 실패 — 원본 유지: %r (%s)", value, col_type)
|
||||
return value
|
||||
|
||||
|
||||
def _restore_dlq_task_types(task: dict) -> dict:
|
||||
"""[M5] DLQ 파일에서 읽은 task 의 payload 타입을 모델 스키마 기준으로 복원."""
|
||||
model_name = _DLQ_TASK_MODELS.get(task.get("type", ""))
|
||||
if not model_name:
|
||||
return task
|
||||
try:
|
||||
from app.models import models as _m
|
||||
model = getattr(_m, model_name, None)
|
||||
if model is None:
|
||||
return task
|
||||
cols = {c.key: c.type for c in model.__table__.columns}
|
||||
except Exception as e:
|
||||
logger.warning("[DLQ] 모델 스키마 조회 실패 (%s): %s", model_name, e)
|
||||
return task
|
||||
|
||||
def _fix(d):
|
||||
if not isinstance(d, dict):
|
||||
return d
|
||||
return {
|
||||
k: (_coerce_value_for_column(v, cols[k]) if k in cols else v)
|
||||
for k, v in d.items()
|
||||
}
|
||||
|
||||
data = task.get("data")
|
||||
if isinstance(data, list):
|
||||
task["data"] = [_fix(row) for row in data]
|
||||
elif isinstance(data, dict):
|
||||
task["data"] = _fix(data)
|
||||
# update 계열은 갱신 필드를 별도 키에 담는다
|
||||
if isinstance(task.get("updates"), dict):
|
||||
task["updates"] = _fix(task["updates"])
|
||||
return task
|
||||
|
||||
|
||||
async def replay_dlq() -> int:
|
||||
"""기동 시 DLQ 파일을 읽어 db_task_queue 에 재큐잉. 재큐잉한 건수 반환.
|
||||
|
||||
[F4 2026-09-28] DLQ 는 지금까지 **쓰기 전용**이었다(읽는 코드 0건). 3회 재시도
|
||||
실패분과 종료 drain 미커밋분이 파일로만 남아 유실이 영구화되고, 복구 수단은
|
||||
수동 psql 뿐이었다.
|
||||
|
||||
CLAUDE.md §3.2 에 따라 **직접 DB 쓰기를 하지 않고 반드시 큐를 경유**한다.
|
||||
파일은 재큐잉 **전에** ``processed/`` 로 옮긴다 — 옮긴 뒤 큐 투입이 실패해도
|
||||
파일은 보존되고(수동 복구 가능), 다음 기동에서 같은 태스크를 두 번 큐에 넣지
|
||||
않는다(insert_trade_history 는 멱등이 아니므로 중복이 더 위험하다).
|
||||
|
||||
손상된 JSON 은 건너뛰고 warning 을 남긴다(파일은 그대로 두어 사람이 본다).
|
||||
"""
|
||||
def _list_files() -> list:
|
||||
if not DLQ_DIR.exists():
|
||||
return []
|
||||
return sorted(f for f in DLQ_DIR.glob("*.json") if f.is_file())
|
||||
|
||||
files = await asyncio.to_thread(_list_files)
|
||||
if not files:
|
||||
return 0
|
||||
|
||||
if len(files) > DLQ_REPLAY_MAX_FILES:
|
||||
logger.error(
|
||||
"❌ [DLQ] 파일 %d건이 상한 %d건을 초과 — 앞의 %d건만 재처리하고 나머지는 남긴다",
|
||||
len(files), DLQ_REPLAY_MAX_FILES, DLQ_REPLAY_MAX_FILES,
|
||||
)
|
||||
files = files[:DLQ_REPLAY_MAX_FILES]
|
||||
|
||||
logger.warning("[DLQ] 재처리 시작: %d건", len(files))
|
||||
requeued = 0
|
||||
skipped = 0
|
||||
for f in files:
|
||||
try:
|
||||
raw = await asyncio.to_thread(f.read_text, encoding="utf-8")
|
||||
entry = json.loads(raw)
|
||||
task = entry.get("task") if isinstance(entry, dict) else None
|
||||
if not isinstance(task, dict) or not task.get("type"):
|
||||
logger.warning("[DLQ] 건너뜀(형식 불량): %s", f.name)
|
||||
skipped += 1
|
||||
continue
|
||||
|
||||
def _move(path=f):
|
||||
DLQ_PROCESSED_DIR.mkdir(parents=True, exist_ok=True)
|
||||
path.rename(DLQ_PROCESSED_DIR / path.name)
|
||||
|
||||
await asyncio.to_thread(_move)
|
||||
# 재시도 카운터를 초기화해 재큐잉분도 3회 재시도를 받게 한다
|
||||
task.pop("_retry_count", None)
|
||||
# [M5] default=str 로 열화된 Decimal/datetime/date 를 되돌린다
|
||||
task = _restore_dlq_task_types(task)
|
||||
await enqueue_db_write(task)
|
||||
requeued += 1
|
||||
except json.JSONDecodeError as e:
|
||||
logger.warning("[DLQ] 건너뜀(JSON 손상): %s — %s", f.name, e)
|
||||
skipped += 1
|
||||
except Exception as e:
|
||||
logger.error("❌ [DLQ] 재처리 실패: %s — %s", f.name, e, exc_info=True)
|
||||
skipped += 1
|
||||
|
||||
logger.warning(
|
||||
"[DLQ] 재처리 완료: 재큐잉 %d건 / 건너뜀 %d건 (처리분은 %s 로 이동)",
|
||||
requeued, skipped, DLQ_PROCESSED_DIR,
|
||||
)
|
||||
return requeued
|
||||
|
||||
|
||||
async def _save_to_dlq(task: dict, error: str) -> None:
|
||||
"""3회 재시도 실패한 task를 DLQ 파일로 저장.
|
||||
|
||||
@@ -381,6 +539,167 @@ async def enqueue_trade_journal(trade, content: str) -> None:
|
||||
})
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# 포지션 추적 계열 핸들러 (공통 종단)
|
||||
#
|
||||
# [F3 2026-09-28] 종료 drain 이 insert_trade_history 만 커밋하고
|
||||
# upsert_portfolio / upsert_active_trade / close_active_trade 는 DLQ 로 보냈다
|
||||
# → "체결 기록은 남는데 포지션 추적 행은 사라지는" 비대칭이 설계에 내장돼 있었다
|
||||
# (재기동 시 restore_from_db 가 복원할 행이 없어 포지션이 고아가 된다).
|
||||
# 정상 경로와 drain 이 **같은 구현**을 쓰도록 여기로 추출한다. 두 곳에 복제하면
|
||||
# 한쪽만 바뀌어 더 나쁜 불일치가 된다(CLAUDE.md §3.3 과 동일한 취지).
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
async def _handle_upsert_portfolio(session, data: dict) -> None:
|
||||
"""Portfolio UPSERT — 정상 경로와 종료 drain 이 공유하는 단일 구현."""
|
||||
from app.models.models import Portfolio
|
||||
|
||||
stock_code = data.get("stock_code")
|
||||
user_id = data.get("user_id")
|
||||
query = select(Portfolio).where(Portfolio.stock_code == stock_code)
|
||||
if user_id is not None:
|
||||
query = query.where(Portfolio.user_id == user_id)
|
||||
result = await session.execute(query)
|
||||
record = result.scalar_one_or_none()
|
||||
|
||||
if record is None:
|
||||
record = Portfolio(**data)
|
||||
session.add(record)
|
||||
else:
|
||||
for key, val in data.items():
|
||||
if key != "stock_code" and val is not None:
|
||||
setattr(record, key, val)
|
||||
|
||||
await session.commit()
|
||||
logger.debug(
|
||||
"✅ [Worker] Portfolio upsert: %s (strategy=%s)",
|
||||
stock_code,
|
||||
data.get("strategy_type", "swing"),
|
||||
)
|
||||
|
||||
|
||||
async def _handle_upsert_active_trade(session, data: dict) -> None:
|
||||
"""ActiveTrade UPSERT — 정상 경로와 종료 drain 이 공유하는 단일 구현."""
|
||||
from app.models.models import ActiveTrade
|
||||
|
||||
stock_code = data.get("stock_code")
|
||||
strategy_name = data.get("strategy_name")
|
||||
user_id = data.get("user_id")
|
||||
|
||||
query = select(ActiveTrade).where(
|
||||
ActiveTrade.stock_code == stock_code,
|
||||
ActiveTrade.status == "active",
|
||||
)
|
||||
# [§8.5 항목6] strategy_name(동등) / strategy_names(IN) /
|
||||
# strategy_name_not(NOT IN + NULL 포함) — 새 키 미사용 시 무변경.
|
||||
query = _apply_strategy_name_filter(query, data, ActiveTrade.strategy_name)
|
||||
if user_id is not None:
|
||||
query = query.where(ActiveTrade.user_id == user_id)
|
||||
# [2026-09-27 §8.5 후속] legacy 경로와 동일하게 '가장 최근 행' 우선.
|
||||
# scalar_one_or_none() 은 한 사용자가 같은 종목에 비-DT active 행을
|
||||
# 2건 이상 보유하면 MultipleResultsFound → 3회 재시도 후 DLQ 로 빠진다.
|
||||
# (멀티테넌트 플래그 ON 전제조건)
|
||||
query = query.order_by(ActiveTrade.id.desc()).limit(1)
|
||||
result = await session.execute(query)
|
||||
record = result.scalars().first()
|
||||
else:
|
||||
# legacy(user_id=None) 경로: 사용자 다중화로 동일 종목 active
|
||||
# 행이 2건 이상 존재할 수 있어 scalar_one_or_none() 이
|
||||
# MultipleResultsFound 를 던질 수 있음 → 가장 최근 행 우선.
|
||||
query = query.order_by(ActiveTrade.id.desc()).limit(1)
|
||||
result = await session.execute(query)
|
||||
record = result.scalars().first()
|
||||
|
||||
if record is None:
|
||||
# remaining_pct_multiply·전략명 필터 키는 신규 생성 시 무시
|
||||
create_data = {
|
||||
k: v for k, v in data.items()
|
||||
if k not in _ACTIVE_TRADE_META_KEYS
|
||||
}
|
||||
# 필수 컬럼(quantity/entry_price) 없이 신규 생성 요청이 오면
|
||||
# IntegrityError로 3회 재시도만 낭비 → 경고 후 스킵
|
||||
_missing = [k for k in ("quantity", "entry_price") if create_data.get(k) is None]
|
||||
if _missing:
|
||||
logger.warning(
|
||||
"⚠️ [Worker] ActiveTrade upsert 스킵: %s active 행 없음 + 필수 키 누락 %s (keys=%s)",
|
||||
stock_code, _missing, sorted(create_data.keys()),
|
||||
)
|
||||
return
|
||||
record = ActiveTrade(**create_data)
|
||||
session.add(record)
|
||||
else:
|
||||
# remaining_pct_multiply: 기존 remaining_pct에 비율을 곱함
|
||||
# (MDD 축소 시 기존 분할매도 이력 반영)
|
||||
# data.get() 사용 — pop()으로 task dict를 파괴적으로 변경하면
|
||||
# 재시도 시 이 키가 사라져 no-op으로 성공 처리되는 버그 방지
|
||||
multiply_ratio = data.get("remaining_pct_multiply")
|
||||
if multiply_ratio is not None:
|
||||
current_remaining = getattr(record, "remaining_pct", 1.0) or 1.0
|
||||
record.remaining_pct = float(current_remaining) * float(multiply_ratio)
|
||||
for key, val in data.items():
|
||||
# [C1 2026-09-28] 반드시 continue — return 이면 'stock_code' 가 모든 payload 의
|
||||
# 첫 키라서 기존 active 행에 대한 모든 upsert 가 첫 이터레이션에서 빠져나가
|
||||
# commit() 에 도달하지 못한다(예외가 없어 성공 처리되고 DLQ 에도 안 남는다).
|
||||
# 파급: 분할매도 잔량 갱신·remaining_pct_multiply·highest_since_entry 트레일링
|
||||
# 갱신이 전부 무효 → 2026-09-23 '4주 보유 중 7주 매도' 재발 경로.
|
||||
if key == "stock_code" or key in _ACTIVE_TRADE_META_KEYS:
|
||||
continue
|
||||
if val is not None:
|
||||
setattr(record, key, val)
|
||||
|
||||
await session.commit()
|
||||
logger.debug(
|
||||
"✅ [Worker] ActiveTrade upsert: %s (strategy=%s)",
|
||||
stock_code,
|
||||
strategy_name,
|
||||
)
|
||||
|
||||
|
||||
async def _handle_close_active_trade(session, data: dict) -> None:
|
||||
"""ActiveTrade close — 정상 경로와 종료 drain 이 공유하는 단일 구현."""
|
||||
from app.models.models import ActiveTrade
|
||||
from datetime import datetime as _dt
|
||||
|
||||
stock_code = data.get("stock_code")
|
||||
strategy_name = data.get("strategy_name")
|
||||
user_id = data.get("user_id")
|
||||
|
||||
query = select(ActiveTrade).where(
|
||||
ActiveTrade.stock_code == stock_code,
|
||||
ActiveTrade.status == "active",
|
||||
)
|
||||
# [§8.5 항목6] strategy_name(동등) / strategy_names(IN) /
|
||||
# strategy_name_not(NOT IN + NULL 포함) — 새 키 미사용 시 무변경.
|
||||
query = _apply_strategy_name_filter(query, data, ActiveTrade.strategy_name)
|
||||
if user_id is not None:
|
||||
query = query.where(ActiveTrade.user_id == user_id)
|
||||
# [2026-09-27 §8.5 후속] legacy 경로와 동일하게 '가장 최근 행' 우선.
|
||||
# scalar_one_or_none() 은 한 사용자가 같은 종목에 비-DT active 행을
|
||||
# 2건 이상 보유하면 MultipleResultsFound → 3회 재시도 후 DLQ 로 빠진다.
|
||||
# (멀티테넌트 플래그 ON 전제조건)
|
||||
query = query.order_by(ActiveTrade.id.desc()).limit(1)
|
||||
result = await session.execute(query)
|
||||
record = result.scalars().first()
|
||||
else:
|
||||
# legacy(user_id=None) 경로: 사용자 다중화로 동일 종목 active
|
||||
# 행이 2건 이상 존재할 수 있어 scalar_one_or_none() 이
|
||||
# MultipleResultsFound 를 던질 수 있음 → 가장 최근 행 우선.
|
||||
query = query.order_by(ActiveTrade.id.desc()).limit(1)
|
||||
result = await session.execute(query)
|
||||
record = result.scalars().first()
|
||||
|
||||
if record:
|
||||
record.status = "closed"
|
||||
record.updated_at = now_kst()
|
||||
await session.commit()
|
||||
logger.debug(
|
||||
"✅ [Worker] ActiveTrade closed: %s (strategy=%s)",
|
||||
stock_code,
|
||||
strategy_name,
|
||||
)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# 워커 본체
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
@@ -579,144 +898,15 @@ async def db_writer_worker() -> None:
|
||||
|
||||
# ── 6. Portfolio UPSERT ────────────────────────────────────────
|
||||
elif task_type == "upsert_portfolio":
|
||||
from app.models.models import Portfolio
|
||||
|
||||
stock_code = data.get("stock_code")
|
||||
user_id = data.get("user_id")
|
||||
query = select(Portfolio).where(Portfolio.stock_code == stock_code)
|
||||
if user_id is not None:
|
||||
query = query.where(Portfolio.user_id == user_id)
|
||||
result = await session.execute(query)
|
||||
record = result.scalar_one_or_none()
|
||||
|
||||
if record is None:
|
||||
record = Portfolio(**data)
|
||||
session.add(record)
|
||||
else:
|
||||
for key, val in data.items():
|
||||
if key != "stock_code" and val is not None:
|
||||
setattr(record, key, val)
|
||||
|
||||
await session.commit()
|
||||
logger.debug(
|
||||
"✅ [Worker] Portfolio upsert: %s (strategy=%s)",
|
||||
stock_code,
|
||||
data.get("strategy_type", "swing"),
|
||||
)
|
||||
await _handle_upsert_portfolio(session, data)
|
||||
|
||||
# ── 7. ActiveTrade UPSERT ──────────────────────────────────────
|
||||
elif task_type == "upsert_active_trade":
|
||||
from app.models.models import ActiveTrade
|
||||
|
||||
stock_code = data.get("stock_code")
|
||||
strategy_name = data.get("strategy_name")
|
||||
user_id = data.get("user_id")
|
||||
|
||||
query = select(ActiveTrade).where(
|
||||
ActiveTrade.stock_code == stock_code,
|
||||
ActiveTrade.status == "active",
|
||||
)
|
||||
# [§8.5 항목6] strategy_name(동등) / strategy_names(IN) /
|
||||
# strategy_name_not(NOT IN + NULL 포함) — 새 키 미사용 시 무변경.
|
||||
query = _apply_strategy_name_filter(query, data, ActiveTrade.strategy_name)
|
||||
if user_id is not None:
|
||||
query = query.where(ActiveTrade.user_id == user_id)
|
||||
# [2026-09-27 §8.5 후속] legacy 경로와 동일하게 '가장 최근 행' 우선.
|
||||
# scalar_one_or_none() 은 한 사용자가 같은 종목에 비-DT active 행을
|
||||
# 2건 이상 보유하면 MultipleResultsFound → 3회 재시도 후 DLQ 로 빠진다.
|
||||
# (멀티테넌트 플래그 ON 전제조건)
|
||||
query = query.order_by(ActiveTrade.id.desc()).limit(1)
|
||||
result = await session.execute(query)
|
||||
record = result.scalars().first()
|
||||
else:
|
||||
# legacy(user_id=None) 경로: 사용자 다중화로 동일 종목 active
|
||||
# 행이 2건 이상 존재할 수 있어 scalar_one_or_none() 이
|
||||
# MultipleResultsFound 를 던질 수 있음 → 가장 최근 행 우선.
|
||||
query = query.order_by(ActiveTrade.id.desc()).limit(1)
|
||||
result = await session.execute(query)
|
||||
record = result.scalars().first()
|
||||
|
||||
if record is None:
|
||||
# remaining_pct_multiply·전략명 필터 키는 신규 생성 시 무시
|
||||
create_data = {
|
||||
k: v for k, v in data.items()
|
||||
if k not in _ACTIVE_TRADE_META_KEYS
|
||||
}
|
||||
# 필수 컬럼(quantity/entry_price) 없이 신규 생성 요청이 오면
|
||||
# IntegrityError로 3회 재시도만 낭비 → 경고 후 스킵
|
||||
_missing = [k for k in ("quantity", "entry_price") if create_data.get(k) is None]
|
||||
if _missing:
|
||||
logger.warning(
|
||||
"⚠️ [Worker] ActiveTrade upsert 스킵: %s active 행 없음 + 필수 키 누락 %s (keys=%s)",
|
||||
stock_code, _missing, sorted(create_data.keys()),
|
||||
)
|
||||
continue
|
||||
record = ActiveTrade(**create_data)
|
||||
session.add(record)
|
||||
else:
|
||||
# remaining_pct_multiply: 기존 remaining_pct에 비율을 곱함
|
||||
# (MDD 축소 시 기존 분할매도 이력 반영)
|
||||
# data.get() 사용 — pop()으로 task dict를 파괴적으로 변경하면
|
||||
# 재시도 시 이 키가 사라져 no-op으로 성공 처리되는 버그 방지
|
||||
multiply_ratio = data.get("remaining_pct_multiply")
|
||||
if multiply_ratio is not None:
|
||||
current_remaining = getattr(record, "remaining_pct", 1.0) or 1.0
|
||||
record.remaining_pct = float(current_remaining) * float(multiply_ratio)
|
||||
for key, val in data.items():
|
||||
if key == "stock_code" or key in _ACTIVE_TRADE_META_KEYS:
|
||||
continue
|
||||
if val is not None:
|
||||
setattr(record, key, val)
|
||||
|
||||
await session.commit()
|
||||
logger.debug(
|
||||
"✅ [Worker] ActiveTrade upsert: %s (strategy=%s)",
|
||||
stock_code,
|
||||
strategy_name,
|
||||
)
|
||||
await _handle_upsert_active_trade(session, data)
|
||||
|
||||
# ── 8. ActiveTrade close ───────────────────────────────────────
|
||||
elif task_type == "close_active_trade":
|
||||
from app.models.models import ActiveTrade
|
||||
from datetime import datetime as _dt
|
||||
|
||||
stock_code = data.get("stock_code")
|
||||
strategy_name = data.get("strategy_name")
|
||||
user_id = data.get("user_id")
|
||||
|
||||
query = select(ActiveTrade).where(
|
||||
ActiveTrade.stock_code == stock_code,
|
||||
ActiveTrade.status == "active",
|
||||
)
|
||||
# [§8.5 항목6] strategy_name(동등) / strategy_names(IN) /
|
||||
# strategy_name_not(NOT IN + NULL 포함) — 새 키 미사용 시 무변경.
|
||||
query = _apply_strategy_name_filter(query, data, ActiveTrade.strategy_name)
|
||||
if user_id is not None:
|
||||
query = query.where(ActiveTrade.user_id == user_id)
|
||||
# [2026-09-27 §8.5 후속] legacy 경로와 동일하게 '가장 최근 행' 우선.
|
||||
# scalar_one_or_none() 은 한 사용자가 같은 종목에 비-DT active 행을
|
||||
# 2건 이상 보유하면 MultipleResultsFound → 3회 재시도 후 DLQ 로 빠진다.
|
||||
# (멀티테넌트 플래그 ON 전제조건)
|
||||
query = query.order_by(ActiveTrade.id.desc()).limit(1)
|
||||
result = await session.execute(query)
|
||||
record = result.scalars().first()
|
||||
else:
|
||||
# legacy(user_id=None) 경로: 사용자 다중화로 동일 종목 active
|
||||
# 행이 2건 이상 존재할 수 있어 scalar_one_or_none() 이
|
||||
# MultipleResultsFound 를 던질 수 있음 → 가장 최근 행 우선.
|
||||
query = query.order_by(ActiveTrade.id.desc()).limit(1)
|
||||
result = await session.execute(query)
|
||||
record = result.scalars().first()
|
||||
|
||||
if record:
|
||||
record.status = "closed"
|
||||
record.updated_at = now_kst()
|
||||
await session.commit()
|
||||
logger.debug(
|
||||
"✅ [Worker] ActiveTrade closed: %s (strategy=%s)",
|
||||
stock_code,
|
||||
strategy_name,
|
||||
)
|
||||
await _handle_close_active_trade(session, data)
|
||||
|
||||
# ── 9. TradeJournal INSERT ────────────────────────────────────
|
||||
elif task_type == "insert_trade_journal":
|
||||
@@ -735,6 +925,7 @@ async def db_writer_worker() -> None:
|
||||
except asyncio.CancelledError:
|
||||
# 큐 drain: 종료 전 남은 태스크 처리하여 데이터 손실 방지
|
||||
remaining = 0
|
||||
_drain_started = time.monotonic()
|
||||
while not queue.empty():
|
||||
try:
|
||||
drain_task = queue.get_nowait()
|
||||
@@ -744,10 +935,35 @@ async def db_writer_worker() -> None:
|
||||
try:
|
||||
drain_type = drain_task.get("type", "unknown")
|
||||
drain_data = drain_task.get("data")
|
||||
# [F3 2026-09-28] 시간 예산 초과분은 커밋을 시도하지 않고 DLQ 로 보낸다.
|
||||
# 종료가 무한정 늘어지면 컨테이너가 SIGKILL 을 받아 전량 유실된다.
|
||||
if (time.monotonic() - _drain_started) > DRAIN_DEADLINE_SECONDS:
|
||||
logger.error(
|
||||
"❌ [DBWorker] Drain 시간 예산 %.0f초 초과 — DLQ 로 보관 (type=%s)",
|
||||
DRAIN_DEADLINE_SECONDS, drain_type,
|
||||
)
|
||||
await _save_to_dlq(drain_task, "shutdown_drain_deadline_exceeded")
|
||||
remaining += 1
|
||||
continue
|
||||
async with AsyncSessionLocal() as drain_session:
|
||||
if drain_type == "insert_trade_history":
|
||||
drain_session.add(TradeHistory(**drain_data))
|
||||
await drain_session.commit()
|
||||
# [F3 2026-09-28] 포지션 추적 계열도 커밋한다.
|
||||
# 이전에는 insert_trade_history 만 커밋하고 아래 3종을 DLQ 로
|
||||
# 보내서 "체결 기록은 남는데 포지션 추적 행은 사라지는" 비대칭이
|
||||
# 있었다 → 재기동 시 restore_from_db 가 복원할 행이 없어
|
||||
# 포지션이 고아가 되고 손절·EOD 청산이 적용되지 않았다.
|
||||
elif drain_type == "upsert_portfolio":
|
||||
await _handle_upsert_portfolio(drain_session, drain_data)
|
||||
elif drain_type == "upsert_active_trade":
|
||||
await _handle_upsert_active_trade(drain_session, drain_data)
|
||||
elif drain_type == "close_active_trade":
|
||||
await _handle_close_active_trade(drain_session, drain_data)
|
||||
elif drain_type == "insert_trade_journal":
|
||||
from app.models.models import TradeJournal
|
||||
drain_session.add(TradeJournal(**drain_data))
|
||||
await drain_session.commit()
|
||||
elif drain_type in (
|
||||
"update_trade_history",
|
||||
"update_trade_history_by_order_id",
|
||||
@@ -755,15 +971,18 @@ async def db_writer_worker() -> None:
|
||||
"bulk_upsert_stock_analysis",
|
||||
"upsert_stock_ban",
|
||||
"upsert_stock_ban_v2",
|
||||
"upsert_portfolio",
|
||||
"upsert_active_trade",
|
||||
"close_active_trade",
|
||||
"insert_trade_journal",
|
||||
):
|
||||
# 단순 재큐잉 대신, 중요 태스크만 커밋 시도
|
||||
# insert_trade_history 외에는 복잡한 로직이므로 DLQ 저장
|
||||
# 분석·밴 계열은 다음 배치/기동 시 재생성되므로 DLQ 보관.
|
||||
# F4 의 replay_dlq() 가 기동 시 재큐잉한다.
|
||||
logger.error(
|
||||
"❌ [DBWorker] Drain 미커밋 → DLQ (type=%s) — 기동 시 replay_dlq() 로 재처리",
|
||||
drain_type,
|
||||
)
|
||||
await _save_to_dlq(drain_task, "shutdown_drain")
|
||||
else:
|
||||
logger.error(
|
||||
"❌ [DBWorker] Drain 미지원 type → DLQ (type=%s)", drain_type,
|
||||
)
|
||||
await _save_to_dlq(drain_task, "shutdown_drain_unknown_type")
|
||||
remaining += 1
|
||||
except Exception as drain_err:
|
||||
|
||||
Reference in New Issue
Block a user