shuim
e4346d30fb
- 장중 10분 주기 journal_intraday_scheduler 신설 (SELECT 전용, 쓰기는 db_worker 큐) - 매도 후 3분 유예(JOURNAL_INTRADAY_GRACE_MIN), trade_id in-flight TTL 30분 - order_id 로 해소된 trade_id 에 일지가 있으면 INSERT skip (중복 방지) - 16:00 배치 enqueue_trade_journal 이 strategy_type 을 넘기지 않아 전 일지가 'unknown' 이던 결함 수정 - analyze_performance profit_factor 를 손익률(%) 합이 아닌 금액 기준으로 (1.22 → 0.62) - 일지 대상 SELL 상태에 PARTIAL 포함 (성과 집계 필터는 불변) - journals 단건 생성: 중복 행 시 500 대신 409, JOURNAL_SELL_STATUSES 사용 - 16:00 배치 조회일수 config 화, 손익 0 기간 PF 오경고 제거 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1319 lines
64 KiB
Python
1319 lines
64 KiB
Python
"""Background DB Writer Worker
|
||
|
||
ARCHITECTURE (CLAUDE.md 3.2):
|
||
- 실시간/콜백 스레드에서는 절대 DB 세션을 직접 열지 않는다.
|
||
- 모든 DB 쓰기 작업은 이 모듈의 db_task_queue 로 위임한다.
|
||
- 워커가 큐를 소비하며 단일 세션에서 순차 처리 → 락 경합 제거.
|
||
|
||
Supported task types:
|
||
- insert_trade_history : dict → TradeHistory INSERT
|
||
- update_trade_history : dict → TradeHistory UPDATE by id (체결 확인 후 슬리피지/수량 반영)
|
||
- update_trade_history_by_order_id : dict → TradeHistory UPDATE by order_id (KIS 주문번호 기준)
|
||
- upsert_stock_analysis : dict → StockAnalysis single UPSERT
|
||
- bulk_upsert_stock_analysis : list → StockAnalysis batch UPSERT
|
||
- upsert_stock_ban : dict → StockBan UPSERT (loss_tracker)
|
||
- insert_news : dict → News INSERT (뉴스 감성 분석 결과 저장)
|
||
- log_mcp_deny : dict → MCP 거부 이벤트 로그 (DB 저장 없이 로그만)
|
||
- upsert_portfolio : dict → Portfolio INSERT/UPDATE (strategy_type 포함)
|
||
- upsert_active_trade : dict → ActiveTrade UPSERT (strategy_name 포함)
|
||
- close_active_trade : dict → ActiveTrade 상태를 'closed'로 변경
|
||
- upsert_stock_ban_v2 : dict → StockBanV2 UPSERT (user_id, stock_code 스코프 — 멀티테넌트)
|
||
|
||
멀티테넌트(사용자 스코프) 확장 (docs/multi_tenant_design.md §3.4):
|
||
- insert_trade_history / update_trade_history(_by_order_id) / upsert_portfolio /
|
||
upsert_active_trade / close_active_trade / insert_trade_journal 는 payload
|
||
data["user_id"] 가 있으면 조회 필터·생성 컬럼에 반영하고, 없으면(None) 기존과
|
||
100% 동일하게 동작한다(하위 호환). 새 task 타입을 추가하지 않는다.
|
||
- stock_bans 는 PK가 stock_code 단일이라 사용자 스코프로 확장 불가 → 신규 테이블
|
||
stock_bans_v2(user_id, stock_code UNIQUE 쌍) + 신규 task upsert_stock_ban_v2 사용.
|
||
"""
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import time
|
||
import uuid
|
||
from datetime import date, datetime
|
||
from pathlib import Path
|
||
|
||
from sqlalchemy import or_, select
|
||
|
||
from app.core.database import AsyncSessionLocal
|
||
from app.models.models import TradeHistory
|
||
from app.core.timeutil import now_kst
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 전략명 상수 / 필터 헬퍼 (CLAUDE.md §8.5 항목 3·6)
|
||
#
|
||
# DT_STRATEGY_NAMES 는 mdd_manager.DT_STRATEGY_NAMES 와 **동일 목록**이어야 한다.
|
||
# 여기(db_worker)를 정본으로 두는 이유:
|
||
# - db_worker 는 어떤 서비스 모듈도 모듈 수준에서 import 하지 않아 순환 참조가 없다.
|
||
# - ActiveTrade 를 실제로 쓰는 계층이므로 필터 표현을 한 곳에 모을 수 있다.
|
||
# mdd_manager 쪽은 담당 에이전트가 달라 이번에 수정하지 않는다(후속 과제).
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
DT_STRATEGY_NAMES: tuple[str, ...] = ("day_trade", "daytrading")
|
||
|
||
|
||
def not_dt_strategy_clause(column):
|
||
"""DT 계열 전략을 제외하는 SQL 3값 논리 안전 필터.
|
||
|
||
`column != "day_trade"` 는 column IS NULL 인 legacy 행까지 함께 탈락시킨다
|
||
(SQL 3값 논리: NULL != 'x' → UNKNOWN → WHERE 에서 거짓 취급).
|
||
→ NULL 행을 명시적으로 포함시킨 뒤 DT 계열만 NOT IN 으로 제외한다.
|
||
|
||
Args:
|
||
column: ActiveTrade.strategy_name 등 비교 대상 컬럼
|
||
|
||
Returns:
|
||
SQLAlchemy boolean expression
|
||
"""
|
||
return or_(column.is_(None), column.notin_(DT_STRATEGY_NAMES))
|
||
|
||
|
||
# upsert_active_trade payload 중 ORM 컬럼이 아닌 '메타' 키 (setattr/생성 인자 제외 대상)
|
||
_ACTIVE_TRADE_META_KEYS = (
|
||
"remaining_pct_multiply",
|
||
"strategy_names",
|
||
"strategy_name_not",
|
||
)
|
||
|
||
|
||
def _apply_strategy_name_filter(query, data: dict, column):
|
||
"""payload 의 전략명 필터 키를 SELECT 쿼리에 적용.
|
||
|
||
지원 키 (우선순위 순 — 상위 키가 있으면 하위 키는 무시):
|
||
1) strategy_name : str → column == value (기존 동작, 무변경)
|
||
2) strategy_names : list → column IN (...)
|
||
3) strategy_name_not : str | list → column IS NULL OR column NOT IN (...)
|
||
|
||
세 키가 모두 없거나 비어 있으면 query 를 **그대로** 반환하므로,
|
||
새 키를 쓰지 않는 기존 호출부의 동작은 수정 전과 완전히 동일하다.
|
||
"""
|
||
strategy_name = data.get("strategy_name")
|
||
if strategy_name:
|
||
return query.where(column == strategy_name)
|
||
|
||
names = data.get("strategy_names")
|
||
if names:
|
||
_names = [n for n in names if n]
|
||
if _names:
|
||
return query.where(column.in_(_names))
|
||
return query
|
||
|
||
excluded = data.get("strategy_name_not")
|
||
if excluded:
|
||
if isinstance(excluded, str):
|
||
excluded = [excluded]
|
||
_excluded = [n for n in excluded if n]
|
||
if _excluded:
|
||
return query.where(or_(column.is_(None), column.notin_(_excluded)))
|
||
return query
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# DLQ (Dead Letter Queue) — 3회 재시도 실패 시 파일로 백업
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
DLQ_DIR = Path.home() / ".cache" / "auto-stock" / "dlq"
|
||
|
||
|
||
# 종료 drain 시간 예산(초). 초과분은 커밋을 포기하고 DLQ 로 보관한다.
|
||
DRAIN_DEADLINE_SECONDS = 20
|
||
# DLQ 재처리 완료 파일 이동 위치 — 재처리 무한루프 방지
|
||
DLQ_PROCESSED_DIR = DLQ_DIR / "processed"
|
||
# [C2 2026-09-30] 나이·충돌로 재큐잉하지 않은 파일 격리 위치 (증거 보존 — 삭제하지 않는다)
|
||
DLQ_EXPIRED_DIR = DLQ_DIR / "expired"
|
||
|
||
# [C2] 나이 제한 — replay_dlq() 가 failed_at 을 읽지 않아 **재기동마다 오래된 상태를
|
||
# 되살렸다.** 예: 9/25 drain 으로 빠진 upsert_active_trade 가 그 사이 전량 청산돼
|
||
# status='closed' 가 된 뒤 9/30 기동에 재큐잉되면, 핸들러가 active 행을 못 찾고
|
||
# 필수 키가 다 있으니 **새 active 행을 INSERT** 한다 → §8.4.1 의 "9/23 유령 행 129개"와
|
||
# 같은 실패 모드가 자동화된다. upsert_portfolio 는 이미 매도된 종목의 Portfolio 행을
|
||
# 재생성해 스윙 청산이 없는 수량을 팔려 한다(주문거부·과다매도).
|
||
#
|
||
# 타입별 차등: 위험의 성질이 다르기 때문이다.
|
||
# - **상태 덮어쓰기**(active_trade/portfolio): payload 가 '그때의 현재 상태'다. 시간이
|
||
# 지나면 진실이 아니라 거짓이 되므로 짧게 잡는다. DT 포지션은 당일 청산(15:20 EOD)이
|
||
# 원칙이라 두 시간이면 이미 다른 국면이다.
|
||
# - **기록 추가**(trade_history/journal/analysis/ban/news): 늦게 들어가도 '없던 기록이
|
||
# 생기는' 것이라 상태를 거짓으로 만들지 않는다. 다만 무한정 허용하면 과거 시각 행이
|
||
# 집계(get_uncovered_trades(days=7)·누적 실현손익)를 오염시키므로 24시간으로 끊는다.
|
||
# 상태 덮어쓰기 계열 — payload 가 '그때의 현재 상태'라 오래되면 거짓이 된다.
|
||
# [M-4 2026-09-30] close_active_trade 를 **여기서 뺐다.** 성격이 다르다: "닫아라"는
|
||
# 지시는 시간이 지나도 거짓이 되지 않는다(이미 닫혀 있으면 no-op, 아직 열려 있으면
|
||
# 닫는 것이 맞다). 오히려 격리하는 쪽이 **반대 방향의 유령 행**을 만든다 — 야간
|
||
# 재기동으로 close 가 격리되면 이미 청산된 포지션의 ActiveTrade 가 active 로 영구
|
||
# 잔존하고(§8.4.1 '유령 행 129개'와 같은 형태), 그 잔량이 다른 안전장치(유입 잔고
|
||
# 격리의 '시스템 보유' 하한)로 흘러든다. portfolio.py 의 정합 루프도 active 행이
|
||
# 2개 이상이면 continue 하므로 자동 복구되지 않는다.
|
||
_DLQ_STATE_TASK_TYPES: tuple[str, ...] = (
|
||
"upsert_active_trade",
|
||
"upsert_portfolio",
|
||
)
|
||
# 현재 상태와의 충돌을 확인할 타입 (나이 분류와 별개 — close 는 충돌 확인만 받는다)
|
||
_DLQ_CONFLICT_CHECK_TYPES: tuple[str, ...] = (
|
||
"upsert_active_trade",
|
||
"upsert_portfolio",
|
||
"close_active_trade",
|
||
)
|
||
# 상태 덮어쓰기 계열: 같은 날(로컬 날짜) + 2시간 이내만 재큐잉
|
||
DLQ_REPLAY_MAX_AGE_STATE_SECONDS: float = 2 * 3600
|
||
# 기록 추가 계열: 24시간 이내만 재큐잉
|
||
DLQ_REPLAY_MAX_AGE_RECORD_SECONDS: float = 24 * 3600
|
||
# [M-4] close_active_trade: 대폭 완화(7일). 무제한으로 두지 않는 이유는 충돌 판정이
|
||
# 불가한 상황(DB 미가동 → M-2 단락)에서 '새 포지션을 닫을' 위험이 남기 때문이다.
|
||
# 충돌 판정이 불가하면 아래에서 기록 계열 상한(24시간)으로 되돌린다.
|
||
DLQ_REPLAY_MAX_AGE_CLOSE_SECONDS: float = 7 * 24 * 3600
|
||
|
||
# [M-2 2026-09-30] 충돌 판정 단락 플래그. replay_dlq() 는 lifespan 에서 await 되므로
|
||
# 기동을 막는다. DB 가 아직 안 뜬 상태에서 파일마다 접속을 재시도하면 기동 지연이
|
||
# (파일 수 × 접속 타임아웃)까지 늘어난다 → 첫 실패 후에는 충돌 판정을 건너뛰고
|
||
# **나이 제한만** 적용한다(나이 제한은 단락 후에도 계속 적용된다).
|
||
_dlq_conflict_check_available: bool = True
|
||
# 한 번의 기동에서 재큐잉할 최대 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
|
||
|
||
|
||
def _dlq_max_age_seconds(task_type: str) -> float:
|
||
"""타입별 재큐잉 허용 나이(초).
|
||
|
||
close_active_trade 는 충돌 판정이 **가능할 때만** 7일까지 허용한다. 판정이 단락된
|
||
상태(DB 미가동)에서는 '새 포지션을 닫을' 위험이 남으므로 기록 계열 상한으로 내린다.
|
||
"""
|
||
if task_type in _DLQ_STATE_TASK_TYPES:
|
||
return DLQ_REPLAY_MAX_AGE_STATE_SECONDS
|
||
if task_type == "close_active_trade":
|
||
if _dlq_conflict_check_available:
|
||
return DLQ_REPLAY_MAX_AGE_CLOSE_SECONDS
|
||
return DLQ_REPLAY_MAX_AGE_RECORD_SECONDS
|
||
return DLQ_REPLAY_MAX_AGE_RECORD_SECONDS
|
||
|
||
|
||
def _dlq_age_verdict(entry: dict, task_type: str) -> tuple[bool, str]:
|
||
"""[C2] failed_at 기준 나이 판정. (재큐잉 허용?, 사유) 반환.
|
||
|
||
``failed_at`` 은 ``_save_to_dlq`` 가 ``datetime.now().isoformat()`` (naive 로컬)로
|
||
쓴다. 같은 시계로 비교해야 하므로 여기서도 ``datetime.now()`` 를 쓴다
|
||
(``now_kst()`` 와 섞으면 컨테이너 TZ 에 따라 9시간 오차가 난다).
|
||
|
||
나이를 알 수 없으면: 상태 덮어쓰기 계열은 **격리**(보수적), 기록 계열은 허용.
|
||
"""
|
||
raw = entry.get("failed_at")
|
||
failed_at = None
|
||
if isinstance(raw, str) and raw:
|
||
try:
|
||
failed_at = datetime.fromisoformat(raw)
|
||
except ValueError:
|
||
failed_at = None
|
||
|
||
if failed_at is None:
|
||
if task_type in _DLQ_STATE_TASK_TYPES:
|
||
return False, f"failed_at 없음/파싱불가({raw!r}) — 상태 덮어쓰기 계열은 보수적 격리"
|
||
# close_active_trade 는 나이를 몰라도 허용한다(닫으라는 지시는 늦어도 유효).
|
||
# 새 포지션을 닫는 위험은 충돌 판정(created_at > failed_at)이 막는다.
|
||
return True, "failed_at 없음 — 기록/close 계열이라 허용"
|
||
|
||
if failed_at.tzinfo is not None:
|
||
failed_at = failed_at.replace(tzinfo=None)
|
||
now = datetime.now()
|
||
age = (now - failed_at).total_seconds()
|
||
limit = _dlq_max_age_seconds(task_type)
|
||
if age > limit:
|
||
return False, (
|
||
f"나이 {age / 3600:.1f}시간 > 상한 {limit / 3600:.1f}시간 "
|
||
f"(failed_at={failed_at.isoformat()})"
|
||
)
|
||
if task_type in _DLQ_STATE_TASK_TYPES and failed_at.date() != now.date():
|
||
# 2시간 이내라도 날짜가 다르면 다른 세션이다(자정 전후 기동)
|
||
return False, (
|
||
f"상태 덮어쓰기 계열인데 날짜가 다름 "
|
||
f"(failed_at={failed_at.date()}, today={now.date()})"
|
||
)
|
||
return True, f"나이 {age / 60:.0f}분"
|
||
|
||
|
||
async def _dlq_state_conflict(task: dict, entry: dict) -> tuple[bool, str]:
|
||
"""[C2] 상태 덮어쓰기 계열이 **현재 상태와 충돌**하는지 읽기 전용으로 확인.
|
||
|
||
나이 제한을 통과해도(≤2시간) 그 사이 청산·재진입이 있었으면 유령 행을 만든다.
|
||
여기서는 SELECT 만 수행하고 ``commit()`` 을 호출하지 않는다 — 기동 시 1회,
|
||
실시간 계층이 아니다(CLAUDE.md §3.2 허용예외③ 와 동일한 성격). **쓰기는 여전히
|
||
큐 경유**다.
|
||
|
||
판정 실패(DB 미가동 등)는 '충돌 아님'으로 돌려 나이 제한만 적용한다 — 여기서
|
||
예외로 죽으면 기동이 막힌다.
|
||
|
||
Returns:
|
||
tuple[bool, str]: (충돌인가?, 사유)
|
||
"""
|
||
global _dlq_conflict_check_available
|
||
|
||
task_type = task.get("type", "")
|
||
data = task.get("data") or {}
|
||
stock_code = data.get("stock_code") if isinstance(data, dict) else None
|
||
if not stock_code or task_type not in _DLQ_CONFLICT_CHECK_TYPES:
|
||
return False, ""
|
||
# [M-2] 이미 판정 불가로 확인됐으면 더 시도하지 않는다(기동 지연 유계화).
|
||
# 로그도 남기지 않는다 — 첫 실패에서 한 번만 남겼다.
|
||
if not _dlq_conflict_check_available:
|
||
return False, ""
|
||
|
||
try:
|
||
from app.models.models import ActiveTrade, Portfolio
|
||
|
||
async with AsyncSessionLocal() as session:
|
||
active = (
|
||
await session.execute(
|
||
select(ActiveTrade)
|
||
.where(
|
||
ActiveTrade.stock_code == stock_code,
|
||
ActiveTrade.status == "active",
|
||
)
|
||
.order_by(ActiveTrade.id.desc())
|
||
.limit(1)
|
||
)
|
||
).scalars().first()
|
||
pf = (
|
||
await session.execute(
|
||
select(Portfolio).where(Portfolio.stock_code == stock_code).limit(1)
|
||
)
|
||
).scalars().first()
|
||
pf_qty = int(getattr(pf, "quantity", 0) or 0) if pf is not None else 0
|
||
|
||
if task_type == "upsert_active_trade":
|
||
# active 행도 없고 보유 수량도 0 → 재큐잉하면 유령 active 행이 생긴다
|
||
if active is None and pf_qty <= 0:
|
||
return True, (
|
||
f"{stock_code}: active 행 없음 + 보유수량 0 → 유령 ActiveTrade 생성 위험"
|
||
)
|
||
elif task_type == "upsert_portfolio":
|
||
# Portfolio 행이 없으면 이미 정리된 종목이다. sync_portfolio() 가 KIS
|
||
# 잔고에서 매 주기 재생성하므로 스킵해도 실보유는 복구된다.
|
||
if pf is None:
|
||
return True, f"{stock_code}: Portfolio 행 없음 → 매도된 종목 재생성 위험"
|
||
elif task_type == "close_active_trade":
|
||
# failed_at 이후에 생긴 active 행이면 '새 포지션'이다 — 그걸 닫으면 안 된다
|
||
raw = entry.get("failed_at")
|
||
try:
|
||
_fa = datetime.fromisoformat(raw) if isinstance(raw, str) else None
|
||
except ValueError:
|
||
_fa = None
|
||
if _fa is not None and _fa.tzinfo is not None:
|
||
_fa = _fa.replace(tzinfo=None)
|
||
_created = getattr(active, "created_at", None) if active is not None else None
|
||
if _created is not None and _fa is not None:
|
||
if _created.tzinfo is not None:
|
||
_created = _created.replace(tzinfo=None)
|
||
if _created > _fa:
|
||
return True, (
|
||
f"{stock_code}: 현재 active 행이 failed_at 이후 생성"
|
||
f"(created_at={_created.isoformat()}) → 새 포지션을 닫을 위험"
|
||
)
|
||
except Exception as e:
|
||
# [M-2] 첫 실패에서 단락한다 — 이후 파일은 접속을 재시도하지 않는다.
|
||
_dlq_conflict_check_available = False
|
||
logger.warning(
|
||
"[DLQ] 상태 충돌 확인 실패 (%s/%s): %s — 이후 파일은 충돌 판정을 건너뛰고 "
|
||
"나이 제한만 적용한다(기동 지연 방지). close_active_trade 허용 나이도 "
|
||
"%.0f시간으로 내려간다.",
|
||
task_type, stock_code, e, DLQ_REPLAY_MAX_AGE_RECORD_SECONDS / 3600,
|
||
)
|
||
return False, ""
|
||
return False, ""
|
||
|
||
|
||
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 을 남긴다(파일은 그대로 두어 사람이 본다).
|
||
|
||
[C2 2026-09-30] **나이 제한 + 상태 충돌 확인**을 거친다. 초기 판은 failed_at 을
|
||
읽지 않아 재기동마다 오래된 상태를 되살렸다(유령 ActiveTrade/Portfolio 생성).
|
||
재큐잉하지 않은 파일은 삭제하지 않고 ``expired/`` 로 격리하고 사유를 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))
|
||
# [M-2] 이번 기동의 충돌 판정 가용성 초기화 (이전 호출의 단락을 물려받지 않는다)
|
||
global _dlq_conflict_check_available
|
||
_dlq_conflict_check_available = True
|
||
requeued = 0
|
||
skipped = 0
|
||
expired = 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
|
||
|
||
task_type = task.get("type", "unknown")
|
||
|
||
def _move_to(dest, path=f):
|
||
dest.mkdir(parents=True, exist_ok=True)
|
||
path.rename(dest / path.name)
|
||
|
||
# [C2] ① 상태 충돌 확인 (읽기 전용) — 유령 행 생성 차단.
|
||
# 나이 판정보다 **먼저** 한다: close_active_trade 의 허용 나이가 '충돌 판정
|
||
# 가능 여부'에 달려 있어(M-4), 판정을 먼저 해야 첫 파일이 blind 로 통과하는
|
||
# 창이 생기지 않는다. DB 장애 시에는 여기서 한 번만 시도하고 단락된다(M-2).
|
||
_conflict, _creason = await _dlq_state_conflict(task, entry)
|
||
if _conflict:
|
||
await asyncio.to_thread(_move_to, DLQ_EXPIRED_DIR)
|
||
logger.warning(
|
||
"[DLQ] 격리(현재 상태와 충돌): %s | type=%s | failed_at=%s | 사유=%s",
|
||
f.name, task_type, entry.get("failed_at"), _creason,
|
||
)
|
||
expired += 1
|
||
continue
|
||
|
||
# [C2] ② 나이 제한 — 오래된 상태 재생이 유실보다 위험하다
|
||
_ok, _why = _dlq_age_verdict(entry, task_type)
|
||
if not _ok:
|
||
await asyncio.to_thread(_move_to, DLQ_EXPIRED_DIR)
|
||
logger.warning(
|
||
"[DLQ] 격리(재큐잉 안 함): %s | type=%s | failed_at=%s | 사유=%s",
|
||
f.name, task_type, entry.get("failed_at"), _why,
|
||
)
|
||
expired += 1
|
||
continue
|
||
|
||
await asyncio.to_thread(_move_to, DLQ_PROCESSED_DIR)
|
||
# 재시도 카운터를 초기화해 재큐잉분도 3회 재시도를 받게 한다
|
||
task.pop("_retry_count", None)
|
||
# [M5] default=str 로 열화된 Decimal/datetime/date 를 되돌린다
|
||
task = _restore_dlq_task_types(task)
|
||
await enqueue_db_write(task)
|
||
logger.info(
|
||
"[DLQ] 재큐잉: %s | type=%s | %s", f.name, task_type, _why,
|
||
)
|
||
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건 / 건너뜀 %d건 "
|
||
"(처리분→%s, 격리분→%s)",
|
||
requeued, expired, skipped, DLQ_PROCESSED_DIR, DLQ_EXPIRED_DIR,
|
||
)
|
||
return requeued
|
||
|
||
|
||
async def _save_to_dlq(task: dict, error: str) -> None:
|
||
"""3회 재시도 실패한 task를 DLQ 파일로 저장.
|
||
|
||
저장 위치: ~/.cache/auto-stock/dlq/{timestamp}_{task_type}_{uuid}.json
|
||
파일 I/O는 asyncio.to_thread로 감싸 이벤트 루프 블로킹을 방지한다.
|
||
"""
|
||
try:
|
||
task_type = task.get("type", "unknown")
|
||
filename = (
|
||
f"{datetime.now().strftime('%Y%m%d_%H%M%S')}"
|
||
f"_{task_type}"
|
||
f"_{uuid.uuid4().hex[:8]}.json"
|
||
)
|
||
dlq_entry = {
|
||
"task": task,
|
||
"error": str(error),
|
||
"failed_at": datetime.now().isoformat(),
|
||
"retry_count": task.get("_retry_count", 0),
|
||
}
|
||
|
||
def _write():
|
||
DLQ_DIR.mkdir(parents=True, exist_ok=True)
|
||
(DLQ_DIR / filename).write_text(
|
||
json.dumps(dlq_entry, default=str, ensure_ascii=False),
|
||
encoding="utf-8",
|
||
)
|
||
|
||
await asyncio.to_thread(_write)
|
||
logger.critical(
|
||
"[DLQ] Task saved to DLQ: %s | type=%s | error=%s",
|
||
filename, task_type, error,
|
||
)
|
||
except Exception as e:
|
||
logger.critical(
|
||
"[DLQ] Failed to save to DLQ: %s | task_type=%s",
|
||
e, task.get("type", "unknown"),
|
||
)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 전역 큐 — 앱 어디서나 import 하여 put() 호출
|
||
# [이슈8 수정] asyncio.Queue()를 모듈 로드 시점에 생성하면 이벤트 루프가 없는
|
||
# 환경(pytest, 스크립트 임포트 등)에서 RuntimeError가 발생할 수 있음.
|
||
# _db_task_queue를 None으로 두고, 첫 접근 시 lazy 초기화하는 패턴으로 변경.
|
||
# 외부 코드는 기존처럼 db_task_queue 이름으로 접근 가능 (get_db_task_queue() 내부 참조).
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
_db_task_queue: asyncio.Queue | None = None
|
||
|
||
|
||
def get_db_task_queue() -> asyncio.Queue:
|
||
"""이벤트 루프 기동 후 최초 호출 시 Queue를 생성(lazy init)하여 반환."""
|
||
global _db_task_queue
|
||
if _db_task_queue is None:
|
||
_db_task_queue = asyncio.Queue()
|
||
return _db_task_queue
|
||
|
||
|
||
# 하위 호환: 기존 코드가 `db_task_queue.put()` 형태로 직접 참조하는 경우를 위해
|
||
# 프록시 객체 대신 get_db_task_queue()를 통해 접근하도록 모듈 수준 alias 제공.
|
||
# 단, 모듈 임포트 시점에 이벤트 루프가 없어도 오류가 발생하지 않음.
|
||
class _QueueProxy:
|
||
"""asyncio.Queue를 lazy init으로 감싸는 투명 프록시."""
|
||
|
||
def __getattr__(self, name):
|
||
return getattr(get_db_task_queue(), name)
|
||
|
||
async def put(self, item):
|
||
await get_db_task_queue().put(item)
|
||
|
||
async def get(self):
|
||
return await get_db_task_queue().get()
|
||
|
||
def task_done(self):
|
||
get_db_task_queue().task_done()
|
||
|
||
def qsize(self):
|
||
return get_db_task_queue().qsize()
|
||
|
||
def empty(self):
|
||
return get_db_task_queue().empty()
|
||
|
||
|
||
db_task_queue: asyncio.Queue = _QueueProxy() # type: ignore[assignment]
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 편의 함수 — 호출 측에서 dict 구성 없이 바로 사용
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
async def enqueue_trade_history(data: dict) -> None:
|
||
"""매매일지 INSERT 를 큐에 등록."""
|
||
await db_task_queue.put({"type": "insert_trade_history", "data": data})
|
||
|
||
|
||
async def enqueue_update_trade_history(trade_id: int, updates: dict) -> None:
|
||
"""기존 TradeHistory 레코드 UPDATE 를 큐에 등록.
|
||
|
||
주로 체결 확인(verify_and_sync) 후 executed_price/quantity/status/slippage 반영 시 사용.
|
||
|
||
Args:
|
||
trade_id: 업데이트할 TradeHistory.id
|
||
updates: 업데이트할 필드 dict (예: {"executed_quantity": 10, "status": "COMPLETED"})
|
||
"""
|
||
await db_task_queue.put({
|
||
"type": "update_trade_history",
|
||
"data": {"id": trade_id, **updates},
|
||
})
|
||
|
||
|
||
async def enqueue_upsert_analysis(row: dict) -> None:
|
||
"""단건 StockAnalysis UPSERT 를 큐에 등록."""
|
||
await db_task_queue.put({"type": "upsert_stock_analysis", "data": row})
|
||
|
||
|
||
async def enqueue_bulk_upsert_analysis(rows: list[dict]) -> None:
|
||
"""다건 StockAnalysis batch UPSERT 를 큐에 등록."""
|
||
if rows:
|
||
await db_task_queue.put({"type": "bulk_upsert_stock_analysis", "data": rows})
|
||
|
||
|
||
async def enqueue_db_write(task: dict) -> None:
|
||
"""범용 DB 쓰기 태스크를 큐에 등록.
|
||
|
||
task 는 반드시 'type' 키를 포함해야 하며, 워커가 지원하는 task type 중 하나여야 한다.
|
||
예: {"type": "log_mcp_deny", "reason": "...", "trade_plan": {...}}
|
||
"""
|
||
await db_task_queue.put(task)
|
||
|
||
|
||
async def enqueue_trade_journal_raw(data: dict) -> None:
|
||
"""TradeJournal INSERT 를 큐에 등록 (순수 dict 전달용)."""
|
||
await db_task_queue.put({"type": "insert_trade_journal", "data": data})
|
||
|
||
|
||
async def enqueue_stock_ban_upsert(stock_code: str, consecutive_losses: int, ban_until) -> None:
|
||
"""StockBan UPSERT 를 큐에 등록 (loss_tracker에서 호출)."""
|
||
await db_task_queue.put({
|
||
"type": "upsert_stock_ban",
|
||
"data": {
|
||
"stock_code": stock_code,
|
||
"consecutive_losses": consecutive_losses,
|
||
"ban_until": ban_until,
|
||
}
|
||
})
|
||
|
||
|
||
async def enqueue_stock_ban_upsert_v2(
|
||
user_id: int, stock_code: str, consecutive_losses: int, ban_until
|
||
) -> None:
|
||
"""[v2] StockBanV2 UPSERT 를 큐에 등록 (사용자 스코프 loss_tracker에서 호출).
|
||
|
||
stock_bans(PK=stock_code, 단일 계정 전제)와 달리 stock_bans_v2 는
|
||
UNIQUE(user_id, stock_code) 이므로 동일 종목을 사용자별로 독립 추적한다.
|
||
"""
|
||
await db_task_queue.put({
|
||
"type": "upsert_stock_ban_v2",
|
||
"data": {
|
||
"user_id": user_id,
|
||
"stock_code": stock_code,
|
||
"consecutive_losses": consecutive_losses,
|
||
"ban_until": ban_until,
|
||
}
|
||
})
|
||
|
||
|
||
async def enqueue_portfolio_upsert(data: dict) -> None:
|
||
"""Portfolio INSERT/UPDATE 를 큐에 등록.
|
||
|
||
data 필수 키: stock_code, quantity, average_price
|
||
data 선택 키: stock_name, current_price, profit_loss, profit_loss_rate,
|
||
market, strategy_type ('swing' | 'daytrading')
|
||
"""
|
||
await db_task_queue.put({"type": "upsert_portfolio", "data": data})
|
||
|
||
|
||
async def enqueue_active_trade_upsert(data: dict) -> None:
|
||
"""ActiveTrade UPSERT 를 큐에 등록.
|
||
|
||
data 필수 키: stock_code, quantity, entry_price
|
||
data 선택 키: strategy_name ('swing' | 'day_trade'), stop_loss,
|
||
take_profit, entry_reason, highest_since_entry
|
||
data 선택 키(조회 필터 전용, ORM 컬럼 아님 — _ACTIVE_TRADE_META_KEYS):
|
||
strategy_names(list → IN),
|
||
strategy_name_not(str|list → NOT IN + strategy_name IS NULL 포함)
|
||
우선순위: strategy_name > strategy_names > strategy_name_not
|
||
"""
|
||
await db_task_queue.put({"type": "upsert_active_trade", "data": data})
|
||
|
||
|
||
async def enqueue_close_active_trade(stock_code: str, strategy_name: str | None = None) -> None:
|
||
"""ActiveTrade 를 'closed' 상태로 변경하는 태스크를 큐에 등록.
|
||
|
||
전략명 부정 조건(strategy_name_not) 등 확장 필터가 필요하면
|
||
시그니처 변경 없이 enqueue_db_write() 로 payload 를 직접 넘긴다:
|
||
await enqueue_db_write({
|
||
"type": "close_active_trade",
|
||
"data": {"stock_code": code, "strategy_name_not": DT_STRATEGY_NAMES},
|
||
})
|
||
"""
|
||
await db_task_queue.put({
|
||
"type": "close_active_trade",
|
||
"data": {"stock_code": stock_code, "strategy_name": strategy_name},
|
||
})
|
||
|
||
|
||
async def enqueue_close_active_trade_v2(
|
||
user_id: int, stock_code: str, strategy_name: str | None = None
|
||
) -> None:
|
||
"""[v2] 사용자 스코프 ActiveTrade 를 'closed' 상태로 변경하는 태스크를 큐에 등록.
|
||
|
||
close_active_trade 와 동일 task type("close_active_trade")을 재사용한다 —
|
||
핸들러가 data.get("user_id")로 필터를 추가하므로 새 타입이 필요 없다.
|
||
"""
|
||
await db_task_queue.put({
|
||
"type": "close_active_trade",
|
||
"data": {"stock_code": stock_code, "strategy_name": strategy_name, "user_id": user_id},
|
||
})
|
||
|
||
|
||
async def enqueue_update_trade_history_by_order_id(order_id: str, updates: dict) -> None:
|
||
"""KIS 주문번호(order_id)로 TradeHistory UPDATE 를 큐에 등록.
|
||
|
||
trade_id 대신 order_id 로 레코드를 조회하여 업데이트한다.
|
||
enqueue_trade_history() 로 INSERT 한 레코드의 id 를 모르는 경우 사용.
|
||
|
||
Args:
|
||
order_id: KIS 주문번호 (TradeHistory.order_id)
|
||
updates: 업데이트할 필드 dict (예: {"executed_quantity": 10, "status": "COMPLETED"})
|
||
"""
|
||
await db_task_queue.put({
|
||
"type": "update_trade_history_by_order_id",
|
||
"data": {"order_id": order_id, **updates},
|
||
})
|
||
|
||
|
||
async def enqueue_trade_journal(trade, content: str) -> None:
|
||
"""TradeJournal INSERT 를 큐에 등록 (main.py post_market_scheduler 용).
|
||
|
||
CLAUDE.md 3.2: 스케줄러(실시간 계층)에서 직접 DB 커밋 금지 → 큐 경유.
|
||
"""
|
||
from datetime import datetime
|
||
from app.services.trade_analyzer import _to_float, _parse_strategy, _parse_exit_type
|
||
|
||
pl_rate = _to_float(trade.profit_loss_rate)
|
||
# [2026-10-01] strategy_type 컬럼(swing/daytrading/manual)을 넘겨야 한다. 빠져 있어
|
||
# entry_reason 키워드만 보다 16:00 배치 일지가 전부 'unknown' 이 됐다
|
||
# (create_journal_record 는 넘기고 있었다 — 같은 거래가 경로에 따라 다른 전략명).
|
||
strategy = _parse_strategy(trade.entry_reason, getattr(trade, "strategy_type", None))
|
||
exit_type = _parse_exit_type(trade.exit_reason)
|
||
mood = "positive" if pl_rate > 0 else ("neutral" if pl_rate > -3.0 else "negative")
|
||
tags = [strategy, exit_type]
|
||
if abs(pl_rate) > 5.0:
|
||
tags.append("significant")
|
||
|
||
await db_task_queue.put({
|
||
"type": "insert_trade_journal",
|
||
"data": {
|
||
"trade_id": trade.id,
|
||
"stock_code": trade.stock_code,
|
||
"stock_name": trade.stock_name,
|
||
"journal_date": trade.executed_at or trade.created_at or now_kst(),
|
||
"content": content,
|
||
"strategy_name": strategy,
|
||
"mood": mood,
|
||
"tags": tags,
|
||
},
|
||
})
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 포지션 추적 계열 핸들러 (공통 종단)
|
||
#
|
||
# [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_insert_trade_journal(session, data: dict) -> None:
|
||
"""TradeJournal INSERT — 정상 경로와 종료 drain 이 공유하는 단일 구현.
|
||
|
||
[2026-09-29 B3] 주문 직후 백그라운드로 만드는 일지는 `TradeHistory.id` 를
|
||
모르기 때문에 `trade_id=0` 을 넣고 있었다. `get_uncovered_trades()` 가
|
||
`TradeJournal.trade_id` 로 커버 판정을 하므로 그 일지는 **아무 거래도 커버하지
|
||
못하고**, 실제 TradeHistory 행이 영구 uncovered 로 남아 16:00 배치가 같은 매도에
|
||
두 번째 일지를 만들었다(실측 중복 13건).
|
||
|
||
호출부는 `order_id`(KIS 주문번호)를 알고 있으므로, 여기(이미 DB 세션 안 —
|
||
CLAUDE.md §3 위반 없음)에서 `order_id → TradeHistory.id` 로 해소한다.
|
||
해소 실패 시 `trade_id` 는 **None** 으로 남는다(0 이 아니다 — 0 은 존재하지 않는
|
||
PK 를 가리키는 거짓 값이다).
|
||
"""
|
||
from app.models.models import TradeJournal
|
||
|
||
payload = dict(data)
|
||
# order_id 는 TradeJournal 컬럼이 아니다 — 해소용 입력이므로 반드시 분리한다.
|
||
order_id = payload.pop("order_id", None)
|
||
|
||
if not payload.get("trade_id") and order_id:
|
||
try:
|
||
result = await session.execute(
|
||
select(TradeHistory.id)
|
||
.where(TradeHistory.order_id == str(order_id))
|
||
.order_by(TradeHistory.id.desc())
|
||
.limit(1)
|
||
)
|
||
resolved = result.scalar_one_or_none()
|
||
if resolved:
|
||
payload["trade_id"] = resolved
|
||
# 해소된 trade_id 로 이미 일지가 있으면 중복 생성 금지 (장중 태스크·16:00 배치와
|
||
# 즉시 생성 경로의 경합). 명시 trade_id 경로는 조회하지 않는다(기존 계약).
|
||
dup = await session.execute(
|
||
select(TradeJournal.id).where(TradeJournal.trade_id == resolved).limit(1)
|
||
)
|
||
if dup.scalar_one_or_none() is not None:
|
||
logger.info(
|
||
"[Worker] TradeJournal skip — trade_id=%s 일지 이미 존재", resolved
|
||
)
|
||
return
|
||
else:
|
||
logger.warning(
|
||
"⚠️ [Worker] TradeJournal trade_id 미해소 (order_id=%s, %s) — "
|
||
"uncovered 판정에서 제외됨",
|
||
order_id, payload.get("stock_code", "?"),
|
||
)
|
||
except Exception as resolve_err:
|
||
logger.warning(
|
||
"⚠️ [Worker] trade_id 해소 실패 (order_id=%s): %s", order_id, resolve_err
|
||
)
|
||
|
||
# 0 은 과거 호출부의 플레이스홀더였다 — 거짓 커버 판정을 만들지 않도록 None 으로.
|
||
if payload.get("trade_id") == 0:
|
||
payload["trade_id"] = None
|
||
|
||
session.add(TradeJournal(**payload))
|
||
await session.commit()
|
||
logger.debug(
|
||
"✅ [Worker] TradeJournal saved: %s (trade_id=%s)",
|
||
payload.get("stock_code", "?"), payload.get("trade_id"),
|
||
)
|
||
|
||
|
||
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,
|
||
)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 워커 본체
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
async def db_writer_worker() -> None:
|
||
"""lifespan 에서 asyncio.create_task() 로 실행되는 영구 백그라운드 워커.
|
||
|
||
큐에서 task 를 꺼내 단일 AsyncSession 안에서 처리한 뒤
|
||
commit / rollback 을 자동 관리한다.
|
||
"""
|
||
logger.info("🛠️ Background DB Writer Worker started")
|
||
queue = get_db_task_queue() # 워커 진입 시 큐 인스턴스 고정
|
||
while True:
|
||
task = await queue.get()
|
||
try:
|
||
task_type = task.get("type")
|
||
data = task.get("data")
|
||
|
||
async with AsyncSessionLocal() as session:
|
||
# ── 1. 매매일지 INSERT ─────────────────────────────────────
|
||
if task_type == "insert_trade_history":
|
||
session.add(TradeHistory(**data))
|
||
await session.commit()
|
||
logger.info(
|
||
"✅ [Worker] TradeHistory saved: %s",
|
||
data.get("stock_code", "?"),
|
||
)
|
||
|
||
# ── 2. TradeHistory UPDATE (체결 확인 후 슬리피지/수량 반영) ──
|
||
elif task_type == "update_trade_history":
|
||
trade_id = data.get("id")
|
||
user_id = data.get("user_id")
|
||
if trade_id:
|
||
query = select(TradeHistory).where(TradeHistory.id == trade_id)
|
||
if user_id is not None:
|
||
query = query.where(TradeHistory.user_id == user_id)
|
||
row = await session.execute(query)
|
||
trade = row.scalar_one_or_none()
|
||
if trade:
|
||
for key, val in data.items():
|
||
if key != "id" and val is not None and hasattr(trade, key):
|
||
setattr(trade, key, val)
|
||
await session.commit()
|
||
logger.info(
|
||
"✅ [Worker] TradeHistory updated: id=%s status=%s",
|
||
trade_id,
|
||
data.get("status", "?"),
|
||
)
|
||
|
||
# ── 2b. TradeHistory UPDATE by order_id (KIS 주문번호 기준) ──
|
||
elif task_type == "update_trade_history_by_order_id":
|
||
order_id = data.get("order_id")
|
||
user_id = data.get("user_id")
|
||
if order_id:
|
||
query = select(TradeHistory).where(TradeHistory.order_id == order_id)
|
||
if user_id is not None:
|
||
query = query.where(TradeHistory.user_id == user_id)
|
||
row = await session.execute(query)
|
||
trade = row.scalar_one_or_none()
|
||
if trade:
|
||
for key, val in data.items():
|
||
if key != "order_id" and val is not None and hasattr(trade, key):
|
||
setattr(trade, key, val)
|
||
await session.commit()
|
||
logger.info(
|
||
"✅ [Worker] TradeHistory updated by order_id=%s status=%s",
|
||
order_id,
|
||
data.get("status", "?"),
|
||
)
|
||
else:
|
||
logger.warning(
|
||
"⚠️ [Worker] TradeHistory not found for order_id=%s",
|
||
order_id,
|
||
)
|
||
|
||
# ── 3. 단건 StockAnalysis UPSERT ──────────────────────────
|
||
elif task_type == "upsert_stock_analysis":
|
||
from app.services.analysis_db_service import _execute_upsert
|
||
await _execute_upsert(session, [data])
|
||
await session.commit()
|
||
logger.debug(
|
||
"✅ [Worker] StockAnalysis upsert: %s",
|
||
data.get("stock_code", "?"),
|
||
)
|
||
|
||
# ── 3. 다건 StockAnalysis batch UPSERT ────────────────────
|
||
elif task_type == "bulk_upsert_stock_analysis":
|
||
from app.services.analysis_db_service import _execute_upsert
|
||
await _execute_upsert(session, data) # data = list[dict]
|
||
await session.commit()
|
||
logger.info(
|
||
"✅ [Worker] Bulk StockAnalysis upsert: %d rows", len(data)
|
||
)
|
||
|
||
# ── 4. StockBan UPSERT ────────────────────────────────────────
|
||
elif task_type == "upsert_stock_ban":
|
||
from app.models.models import StockBan
|
||
from app.core.dt_utils import to_naive_utc
|
||
|
||
stock_code = data.get("stock_code")
|
||
consecutive_losses = data.get("consecutive_losses")
|
||
# stock_bans.ban_until은 tz-naive 컬럼 — tz-aware 값이 들어와도 안전하게 저장.
|
||
ban_until = to_naive_utc(data.get("ban_until"))
|
||
|
||
result = await session.execute(select(StockBan).where(StockBan.stock_code == stock_code))
|
||
record = result.scalar_one_or_none()
|
||
|
||
if record is None:
|
||
record = StockBan(
|
||
stock_code=stock_code,
|
||
consecutive_losses=consecutive_losses,
|
||
ban_until=ban_until,
|
||
)
|
||
session.add(record)
|
||
else:
|
||
record.consecutive_losses = consecutive_losses
|
||
record.ban_until = ban_until
|
||
|
||
await session.commit()
|
||
logger.debug(
|
||
"✅ [Worker] StockBan upsert: %s (losses=%d, ban_until=%s)",
|
||
stock_code,
|
||
consecutive_losses,
|
||
ban_until.isoformat() if ban_until else "None",
|
||
)
|
||
|
||
# ── 4v2. StockBanV2 UPSERT (user_id, stock_code 스코프) ────────
|
||
elif task_type == "upsert_stock_ban_v2":
|
||
from app.models.models import StockBanV2
|
||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||
from app.core.dt_utils import to_naive_utc
|
||
|
||
user_id = data.get("user_id")
|
||
stock_code = data.get("stock_code")
|
||
consecutive_losses = data.get("consecutive_losses")
|
||
# stock_bans_v2.ban_until도 tz-naive 컬럼 — tz-aware 값이 들어와도 안전하게 저장.
|
||
ban_until = to_naive_utc(data.get("ban_until"))
|
||
|
||
# PostgreSQL INSERT ... ON CONFLICT DO UPDATE — 여러 워커/사용자
|
||
# 사이클이 동시에 같은 (user_id, stock_code)를 갱신해도 경쟁 안전.
|
||
stmt = pg_insert(StockBanV2).values(
|
||
user_id=user_id,
|
||
stock_code=stock_code,
|
||
consecutive_losses=consecutive_losses,
|
||
ban_until=ban_until,
|
||
)
|
||
stmt = stmt.on_conflict_do_update(
|
||
constraint="uq_stock_bans_v2_user_stock",
|
||
set_={
|
||
"consecutive_losses": stmt.excluded.consecutive_losses,
|
||
"ban_until": stmt.excluded.ban_until,
|
||
"updated_at": now_kst(),
|
||
},
|
||
)
|
||
await session.execute(stmt)
|
||
await session.commit()
|
||
logger.debug(
|
||
"✅ [Worker] StockBanV2 upsert: user=%s %s (losses=%s, ban_until=%s)",
|
||
user_id, stock_code, consecutive_losses,
|
||
ban_until.isoformat() if ban_until else "None",
|
||
)
|
||
|
||
# ── 4b. News INSERT (중복 URL 무시) ────────────────────────────
|
||
elif task_type == "insert_news":
|
||
from app.models.models import News
|
||
|
||
news_url = data.get("url")
|
||
is_dup = False
|
||
# URL 기반 중복 방지 (동일 URL은 skip)
|
||
if news_url:
|
||
dup_result = await session.execute(
|
||
select(News.id).where(News.url == news_url).limit(1)
|
||
)
|
||
if dup_result.scalar_one_or_none() is not None:
|
||
is_dup = True
|
||
logger.debug(
|
||
"⏭️ [Worker] News skip (dup URL): %s",
|
||
(data.get("title") or "?")[:40],
|
||
)
|
||
|
||
if not is_dup:
|
||
session.add(News(**data))
|
||
await session.commit()
|
||
logger.debug(
|
||
"✅ [Worker] News saved: %s",
|
||
(data.get("title") or "?")[:40],
|
||
)
|
||
|
||
# ── 5. MCP 거부 이벤트 로그 ───────────────────────────────────
|
||
elif task_type == "log_mcp_deny":
|
||
# DB 저장 없이 로그만 기록 (필요 시 별도 테이블 확장 가능)
|
||
logger.warning(
|
||
"🛑 [Worker] MCP DENY logged: reason=%s, trade_plan=%s",
|
||
task.get("reason", "unknown"),
|
||
task.get("trade_plan"),
|
||
)
|
||
|
||
# ── 6. Portfolio UPSERT ────────────────────────────────────────
|
||
elif task_type == "upsert_portfolio":
|
||
await _handle_upsert_portfolio(session, data)
|
||
|
||
# ── 7. ActiveTrade UPSERT ──────────────────────────────────────
|
||
elif task_type == "upsert_active_trade":
|
||
await _handle_upsert_active_trade(session, data)
|
||
|
||
# ── 8. ActiveTrade close ───────────────────────────────────────
|
||
elif task_type == "close_active_trade":
|
||
await _handle_close_active_trade(session, data)
|
||
|
||
# ── 9. TradeJournal INSERT ────────────────────────────────────
|
||
elif task_type == "insert_trade_journal":
|
||
await _handle_insert_trade_journal(session, data)
|
||
|
||
else:
|
||
logger.warning("⚠️ [Worker] Unknown task type: %s", task_type)
|
||
|
||
except asyncio.CancelledError:
|
||
# 큐 drain: 종료 전 남은 태스크 처리하여 데이터 손실 방지
|
||
remaining = 0
|
||
_drain_started = time.monotonic()
|
||
while not queue.empty():
|
||
try:
|
||
drain_task = queue.get_nowait()
|
||
except asyncio.QueueEmpty:
|
||
# get_nowait()이 실패해 아이템을 꺼내지 못했으므로 task_done() 호출 금지
|
||
break
|
||
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":
|
||
await _handle_insert_trade_journal(drain_session, drain_data)
|
||
elif drain_type in (
|
||
"update_trade_history",
|
||
"update_trade_history_by_order_id",
|
||
"upsert_stock_analysis",
|
||
"bulk_upsert_stock_analysis",
|
||
"upsert_stock_ban",
|
||
"upsert_stock_ban_v2",
|
||
):
|
||
# 분석·밴 계열은 다음 배치/기동 시 재생성되므로 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:
|
||
logger.error(f"[DBWorker] Drain 실패: {drain_err}")
|
||
try:
|
||
await _save_to_dlq(drain_task, str(drain_err))
|
||
except Exception:
|
||
pass
|
||
remaining += 1
|
||
finally:
|
||
queue.task_done()
|
||
if remaining:
|
||
logger.info(f"[DBWorker] 종료 시 {remaining}건 drain 완료")
|
||
logger.info("⏹️ DB Writer Worker stopped")
|
||
break
|
||
except Exception as exc:
|
||
retry_count = task.get("_retry_count", 0)
|
||
max_retries = 3
|
||
task_type = task.get("type", "unknown")
|
||
|
||
if retry_count < max_retries:
|
||
task["_retry_count"] = retry_count + 1
|
||
delay = min(2 ** retry_count, 8) # 1, 2, 4초 지수 백오프
|
||
logger.warning(
|
||
"⚠️ [Worker] DB task retry %d/%d (type=%s): %s — %ds 후 재시도",
|
||
retry_count + 1, max_retries, task_type, exc, delay,
|
||
)
|
||
await asyncio.sleep(delay)
|
||
await queue.put(task) # 큐에 재등록
|
||
else:
|
||
logger.error(
|
||
"❌ [Worker] DB task DROPPED after %d retries (type=%s): %s",
|
||
max_retries, task_type, exc, exc_info=True,
|
||
)
|
||
# DLQ: 최종 실패 task를 파일로 백업하여 데이터 손실 방지
|
||
await _save_to_dlq(task, str(exc))
|
||
finally:
|
||
queue.task_done()
|