fix(journal): 장중 일지 자동 생성 + 전략명 unknown + profit_factor 단위 오류
- 장중 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>
This commit is contained in:
@@ -754,7 +754,10 @@ async def enqueue_trade_journal(trade, content: str) -> None:
|
||||
from app.services.trade_analyzer import _to_float, _parse_strategy, _parse_exit_type
|
||||
|
||||
pl_rate = _to_float(trade.profit_loss_rate)
|
||||
strategy = _parse_strategy(trade.entry_reason)
|
||||
# [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]
|
||||
@@ -819,6 +822,16 @@ async def _handle_insert_trade_journal(session, data: dict) -> None:
|
||||
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) — "
|
||||
|
||||
@@ -247,6 +247,85 @@ async def batch_analysis_scheduler():
|
||||
raise
|
||||
|
||||
|
||||
# ── 매매일지 큐 등록 (16:00 배치·장중 주기 공용) ─────────────────────────────
|
||||
# db_worker 큐는 비동기라 enqueue → INSERT 커밋 사이에 같은 uncovered 조회가 다시 돌면
|
||||
# 같은 trade_id 를 또 큐에 넣는다. 프로세스 내 in-flight 집합으로 막는다
|
||||
# (재시작 시에는 DB 의 trade_id 커버 판정이 단일 진실). 무한 증가 방지로 상한을 둔다.
|
||||
_JOURNAL_ENQUEUED_IDS: dict[int, float] = {} # trade_id -> 등록 monotonic 시각(TTL 만료)
|
||||
_JOURNAL_ENQUEUED_MAX = 5000
|
||||
|
||||
|
||||
async def _enqueue_journals_v2(trades, perf=None) -> int:
|
||||
"""uncovered SELL 거래를 일지로 만들어 db_worker 큐에 등록한다. 등록 건수 반환.
|
||||
|
||||
멱등: 이미 이번 프로세스에서 등록한 trade_id 는 건너뛴다. DB 쓰기는 큐 경유(§3.2).
|
||||
"""
|
||||
from app.services.trade_analyzer import generate_journal_for_trade
|
||||
from app.services.db_worker import enqueue_trade_journal
|
||||
|
||||
import time as _time
|
||||
|
||||
_now_m = _time.monotonic()
|
||||
_ttl = max(1, int(settings.JOURNAL_ENQUEUE_TTL_MIN)) * 60
|
||||
for _k in [k for k, ts in _JOURNAL_ENQUEUED_IDS.items() if _now_m - ts > _ttl]:
|
||||
del _JOURNAL_ENQUEUED_IDS[_k]
|
||||
if len(_JOURNAL_ENQUEUED_IDS) > _JOURNAL_ENQUEUED_MAX:
|
||||
_JOURNAL_ENQUEUED_IDS.clear()
|
||||
|
||||
count = 0
|
||||
for trade in trades:
|
||||
tid = getattr(trade, "id", None)
|
||||
if tid is not None and tid in _JOURNAL_ENQUEUED_IDS:
|
||||
continue
|
||||
try:
|
||||
content = generate_journal_for_trade(trade, perf)
|
||||
await enqueue_trade_journal(trade, content)
|
||||
if tid is not None:
|
||||
_JOURNAL_ENQUEUED_IDS[tid] = _now_m
|
||||
count += 1
|
||||
except Exception as e:
|
||||
logger.warning(f"일지 생성 실패 ({getattr(trade, 'stock_code', '?')}): {e}")
|
||||
return count
|
||||
|
||||
|
||||
async def journal_intraday_scheduler():
|
||||
"""장중 N분 주기로 uncovered SELL 을 일지화한다 (16:00 배치의 장중 보충).
|
||||
|
||||
별도 배치 태스크라 실시간 계층(체결 콜백)에서 DB I/O 를 하지 않는다. 읽기만 하고
|
||||
쓰기는 db_worker 큐 경유. 멱등성은 trade_id 커버 판정 + in-flight 집합이 보장한다.
|
||||
"""
|
||||
await asyncio.sleep(90) # 부팅 안정화
|
||||
logger.info(
|
||||
"📝 Intraday journal scheduler started (every %s min)",
|
||||
settings.JOURNAL_INTRADAY_INTERVAL_MIN,
|
||||
)
|
||||
try:
|
||||
while True:
|
||||
interval = max(1, int(settings.JOURNAL_INTRADAY_INTERVAL_MIN)) * 60
|
||||
if settings.JOURNAL_INTRADAY_ENABLED and is_market_open():
|
||||
try:
|
||||
from app.services.trade_analyzer import get_uncovered_trades_v2
|
||||
|
||||
async with AsyncSessionLocal() as db: # SELECT 전용, commit 없음
|
||||
uncovered = await get_uncovered_trades_v2(
|
||||
db,
|
||||
days=settings.JOURNAL_LOOKBACK_DAYS,
|
||||
min_age_minutes=settings.JOURNAL_INTRADAY_GRACE_MIN,
|
||||
)
|
||||
n = await _enqueue_journals_v2(uncovered, None)
|
||||
if n:
|
||||
logger.info("📝 [IntradayJournal] 일지 %d건 큐 등록", n)
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.warning("⚠️ [IntradayJournal] 실행 실패: %s", e, exc_info=True)
|
||||
await asyncio.sleep(interval)
|
||||
except asyncio.CancelledError:
|
||||
logger.info("⏹️ Intraday journal scheduler stopped")
|
||||
raise
|
||||
|
||||
|
||||
|
||||
async def post_market_analyzer_scheduler():
|
||||
"""장 종료 후 16:00 KST 1회 실행 — 매매일지 자동 생성 + 전략 자동 최적화.
|
||||
|
||||
@@ -300,18 +379,9 @@ async def post_market_analyzer_scheduler():
|
||||
from sqlalchemy import select, func, and_
|
||||
|
||||
async with AsyncSessionLocal() as db:
|
||||
uncovered = await get_uncovered_trades(db, days=7)
|
||||
uncovered = await get_uncovered_trades(db, days=settings.JOURNAL_LOOKBACK_DAYS)
|
||||
perf = await analyze_performance(db, days=30)
|
||||
journal_count = 0
|
||||
for trade in uncovered:
|
||||
try:
|
||||
content = generate_journal_for_trade(trade, perf)
|
||||
# [FIX] CLAUDE.md 3.2: 스케줄러에서 직접 DB 쓰기 금지 → db_worker 큐 경유
|
||||
from app.services.db_worker import enqueue_trade_journal
|
||||
await enqueue_trade_journal(trade, content)
|
||||
journal_count += 1
|
||||
except Exception as e:
|
||||
logger.warning(f"일지 생성 실패 ({trade.stock_code}): {e}")
|
||||
journal_count = await _enqueue_journals_v2(uncovered, perf)
|
||||
if journal_count > 0:
|
||||
logger.info(f"📝 [PostMarket] 일지 {journal_count}건 큐 등록 완료")
|
||||
|
||||
|
||||
@@ -455,13 +455,19 @@ async def analyze_performance(db: AsyncSession, days: int = 30) -> Dict[str, Any
|
||||
win_rate = len(wins) / len(pl_rates) if pl_rates else 0.0
|
||||
avg_win = sum(wins) / len(wins) if wins else 0.0
|
||||
avg_loss = sum(losses) / len(losses) if losses else 0.0
|
||||
profit_factor = (
|
||||
(sum(wins) / abs(sum(losses))) if losses and sum(losses) != 0 else float("inf")
|
||||
)
|
||||
|
||||
gross_profit = sum(a for a in pl_amounts if a > 0)
|
||||
gross_loss = sum(a for a in pl_amounts if a < 0)
|
||||
net_pnl = gross_profit + gross_loss
|
||||
# [2026-10-01] PF 는 **금액** 기준이다(gross_profit / |gross_loss|). 이전에는 손익률(%)
|
||||
# 합계로 계산해 gross_profit 266,370 / gross_loss -430,619 인데 PF 1.2165 가 나왔다
|
||||
# (소액 거래의 큰 %가 대금 큰 거래의 작은 %를 압도). 손실 없음 → inf(아래서 99 로 캡),
|
||||
# 수익도 손실도 없음 → 0.
|
||||
if gross_loss < 0:
|
||||
profit_factor = gross_profit / abs(gross_loss)
|
||||
elif gross_profit > 0:
|
||||
profit_factor = float("inf")
|
||||
else:
|
||||
profit_factor = 0.0
|
||||
|
||||
# 최대 연속 손실
|
||||
max_consecutive_loss = _calc_max_consecutive_loss(pl_rates)
|
||||
@@ -636,7 +642,11 @@ def build_insights(perf: Dict[str, Any]) -> List[Dict[str, str]]:
|
||||
else:
|
||||
insights.append({"level": "info", "message": f"승률 {win_rate:.1f}% — 양호합니다."})
|
||||
|
||||
if pf < 1.0:
|
||||
# 거래 없음/수익·손실 모두 0 이면 PF=0 은 "손실>수익"이 아니라 판단 불가 — 인사이트 생략
|
||||
_has_pnl = bool(summary.get("gross_profit") or summary.get("gross_loss"))
|
||||
if not _has_pnl:
|
||||
pass
|
||||
elif pf < 1.0:
|
||||
insights.append({"level": "danger", "message": f"Profit Factor {pf:.2f} — 손실이 수익보다 큽니다. 손절 기준 강화 필요."})
|
||||
elif pf < 1.5:
|
||||
insights.append({"level": "warning", "message": f"Profit Factor {pf:.2f} — 개선 여지 있습니다."})
|
||||
@@ -688,9 +698,29 @@ def build_insights(perf: Dict[str, Any]) -> List[Dict[str, str]]:
|
||||
# 매매일지 생성
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
# 일지 대상 SELL 상태. 성과 집계(analyze_performance)의 COMPLETED/FILLED 와 **의도적으로 다르다**:
|
||||
# PARTIAL 은 실제 체결된 매도(부분 체결)라 기록할 가치가 있고, 이후 COMPLETED 로 갱신돼도
|
||||
# 같은 TradeHistory.id 라 trade_id 커버 판정으로 중복되지 않는다. PENDING(체결 미확인)·
|
||||
# FAILED(거부)는 손익이 확정되지 않았거나 거래가 없으므로 제외한다.
|
||||
# 성과 집계에는 PARTIAL 을 넣지 않는다(손익 이중 계상 방지) — 이 상수는 일지 전용.
|
||||
JOURNAL_SELL_STATUSES = ("COMPLETED", "FILLED", "PARTIAL")
|
||||
|
||||
|
||||
async def get_uncovered_trades(db: AsyncSession, days: int = 7) -> List[TradeHistory]:
|
||||
"""일지가 아직 없는 완료 SELL 거래 목록을 반환."""
|
||||
"""일지가 아직 없는 체결 SELL(COMPLETED/FILLED/PARTIAL) 거래 목록을 반환."""
|
||||
return await get_uncovered_trades_v2(db, days=days, min_age_minutes=0)
|
||||
|
||||
|
||||
async def get_uncovered_trades_v2(
|
||||
db: AsyncSession, days: int = 7, min_age_minutes: int = 0
|
||||
) -> List[TradeHistory]:
|
||||
"""get_uncovered_trades + 유예시간: created_at 이 min_age_minutes 이내인 SELL 은 제외.
|
||||
|
||||
매도 직후 `_create_journal_background`(MCP 분석 최대 30초 후 큐 등록)가 같은 매도의
|
||||
일지를 만들 시간을 준다 — 장중 주기 태스크가 그 사이에 중복 생성하지 않도록.
|
||||
"""
|
||||
since = now_kst() - timedelta(days=days)
|
||||
cutoff = now_kst() - timedelta(minutes=max(0, int(min_age_minutes)))
|
||||
|
||||
# 이미 일지가 있는 trade_id 목록
|
||||
# [2026-09-29 B3] `> 0` 조건이 필수다. 과거 호출부가 trade_id 미확보 시 `0` 을
|
||||
@@ -703,8 +733,9 @@ async def get_uncovered_trades(db: AsyncSession, days: int = 7) -> List[TradeHis
|
||||
stmt = select(TradeHistory).where(
|
||||
and_(
|
||||
TradeHistory.order_type == "SELL",
|
||||
TradeHistory.status.in_(["COMPLETED", "FILLED"]),
|
||||
TradeHistory.status.in_(JOURNAL_SELL_STATUSES),
|
||||
TradeHistory.created_at >= since,
|
||||
TradeHistory.created_at <= cutoff,
|
||||
TradeHistory.id.not_in(covered_ids) if covered_ids else True,
|
||||
)
|
||||
).order_by(TradeHistory.created_at.desc())
|
||||
|
||||
Reference in New Issue
Block a user