Compare commits
2 Commits
97f77d568c
...
4fa259a27f
| Author | SHA1 | Date | |
|---|---|---|---|
| 4fa259a27f | |||
| e4346d30fb |
@@ -131,12 +131,13 @@ async def generate_journal_for_trade_id(
|
|||||||
analyze_performance,
|
analyze_performance,
|
||||||
generate_journal_for_trade,
|
generate_journal_for_trade,
|
||||||
create_journal_record,
|
create_journal_record,
|
||||||
|
JOURNAL_SELL_STATUSES,
|
||||||
)
|
)
|
||||||
# 기존 일지 확인
|
# 기존 일지 확인
|
||||||
existing = await db.execute(
|
existing = await db.execute(
|
||||||
select(TradeJournal).where(TradeJournal.trade_id == trade_id)
|
select(TradeJournal).where(TradeJournal.trade_id == trade_id)
|
||||||
)
|
)
|
||||||
if existing.scalar_one_or_none():
|
if existing.scalars().first(): # 중복 행이 있어도 500 이 아니라 409
|
||||||
raise HTTPException(status_code=409, detail="이미 일지가 존재합니다.")
|
raise HTTPException(status_code=409, detail="이미 일지가 존재합니다.")
|
||||||
|
|
||||||
# 거래 조회
|
# 거래 조회
|
||||||
@@ -146,7 +147,7 @@ async def generate_journal_for_trade_id(
|
|||||||
trade = trade_result.scalar_one_or_none()
|
trade = trade_result.scalar_one_or_none()
|
||||||
if not trade:
|
if not trade:
|
||||||
raise HTTPException(status_code=404, detail="거래 이력을 찾을 수 없습니다.")
|
raise HTTPException(status_code=404, detail="거래 이력을 찾을 수 없습니다.")
|
||||||
if trade.order_type != "SELL" or trade.status not in ("COMPLETED", "FILLED"):
|
if trade.order_type != "SELL" or trade.status not in JOURNAL_SELL_STATUSES:
|
||||||
raise HTTPException(status_code=400, detail="완료된 매도 거래에 대해서만 일지를 생성할 수 있습니다.")
|
raise HTTPException(status_code=400, detail="완료된 매도 거래에 대해서만 일지를 생성할 수 있습니다.")
|
||||||
|
|
||||||
perf = await analyze_performance(db, days=30)
|
perf = await analyze_performance(db, days=30)
|
||||||
|
|||||||
@@ -356,6 +356,18 @@ class Settings(BaseSettings):
|
|||||||
# 남긴다 — 적용은 운영자가 전략설정 UI 에서 수동으로 한다.
|
# 남긴다 — 적용은 운영자가 전략설정 UI 에서 수동으로 한다.
|
||||||
STRATEGY_AUTO_APPLY_ENABLED: bool = False
|
STRATEGY_AUTO_APPLY_ENABLED: bool = False
|
||||||
|
|
||||||
|
# 장중 매매일지 보충 생성 — 체결된 SELL 중 일지 없는 건을 N분 주기로 일지화한다.
|
||||||
|
# (16:00 배치만 있으면 장중 매도는 최대 6시간 뒤에야 일지가 생긴다. 2026-09-28~29 에
|
||||||
|
# 장중 일지가 있었던 것은 스케줄러가 아니라 수동 "일지 생성" 버튼 호출 때문이었다.)
|
||||||
|
JOURNAL_INTRADAY_ENABLED: bool = True
|
||||||
|
JOURNAL_INTRADAY_INTERVAL_MIN: int = 10
|
||||||
|
# 한 번에 훑는 과거 일수 (uncovered 판정은 trade_id 기준이라 멱등)
|
||||||
|
JOURNAL_LOOKBACK_DAYS: int = 7
|
||||||
|
# 장중 태스크는 created_at 이 이 시간(분) 이내인 SELL 을 건너뛴다 (즉시 생성 경로와의 경합 방지)
|
||||||
|
JOURNAL_INTRADAY_GRACE_MIN: int = 3
|
||||||
|
# in-flight 등록 기록 만료(분) — 큐 INSERT 가 DLQ 로 갔어도 이 시간 뒤엔 재시도된다
|
||||||
|
JOURNAL_ENQUEUE_TTL_MIN: int = 30
|
||||||
|
|
||||||
# ═════════════════════════════════════════════════════════════════════════════
|
# ═════════════════════════════════════════════════════════════════════════════
|
||||||
# ⚠️ **백테스트 미검증 — True 로 바꾸지 말 것 (CLAUDE.md §8.1-4).**
|
# ⚠️ **백테스트 미검증 — True 로 바꾸지 말 것 (CLAUDE.md §8.1-4).**
|
||||||
# 이 전략은 SIDEWAYS 국면 전용이며, market_analyzer 의 SIDEWAYS 전면 차단(§6.1)을
|
# 이 전략은 SIDEWAYS 국면 전용이며, market_analyzer 의 SIDEWAYS 전면 차단(§6.1)을
|
||||||
|
|||||||
@@ -57,6 +57,7 @@ logger = logging.getLogger(__name__)
|
|||||||
from app.services.schedulers import (
|
from app.services.schedulers import (
|
||||||
batch_analysis_scheduler,
|
batch_analysis_scheduler,
|
||||||
post_market_analyzer_scheduler,
|
post_market_analyzer_scheduler,
|
||||||
|
journal_intraday_scheduler,
|
||||||
krx_data_collector_scheduler,
|
krx_data_collector_scheduler,
|
||||||
news_collect_scheduler,
|
news_collect_scheduler,
|
||||||
pre_market_us_scheduler,
|
pre_market_us_scheduler,
|
||||||
@@ -280,6 +281,7 @@ async def lifespan(app: FastAPI):
|
|||||||
app.state.batch_analysis_task = asyncio.create_task(batch_analysis_scheduler())
|
app.state.batch_analysis_task = asyncio.create_task(batch_analysis_scheduler())
|
||||||
app.state.krx_collector_task = asyncio.create_task(krx_data_collector_scheduler())
|
app.state.krx_collector_task = asyncio.create_task(krx_data_collector_scheduler())
|
||||||
app.state.post_market_task = asyncio.create_task(post_market_analyzer_scheduler())
|
app.state.post_market_task = asyncio.create_task(post_market_analyzer_scheduler())
|
||||||
|
app.state.journal_intraday_task = asyncio.create_task(journal_intraday_scheduler())
|
||||||
|
|
||||||
# ── 미국장 자동매매 스케줄러 (US_TRADING_ENABLED=True 시 활성화) ──────────
|
# ── 미국장 자동매매 스케줄러 (US_TRADING_ENABLED=True 시 활성화) ──────────
|
||||||
if getattr(settings, 'US_TRADING_ENABLED', False):
|
if getattr(settings, 'US_TRADING_ENABLED', False):
|
||||||
@@ -424,6 +426,7 @@ async def lifespan(app: FastAPI):
|
|||||||
"day_trade_task",
|
"day_trade_task",
|
||||||
"krx_collector_task",
|
"krx_collector_task",
|
||||||
"post_market_task",
|
"post_market_task",
|
||||||
|
"journal_intraday_task",
|
||||||
"db_worker_task",
|
"db_worker_task",
|
||||||
"kis_stream_task",
|
"kis_stream_task",
|
||||||
"us_trade_task",
|
"us_trade_task",
|
||||||
|
|||||||
@@ -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
|
from app.services.trade_analyzer import _to_float, _parse_strategy, _parse_exit_type
|
||||||
|
|
||||||
pl_rate = _to_float(trade.profit_loss_rate)
|
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)
|
exit_type = _parse_exit_type(trade.exit_reason)
|
||||||
mood = "positive" if pl_rate > 0 else ("neutral" if pl_rate > -3.0 else "negative")
|
mood = "positive" if pl_rate > 0 else ("neutral" if pl_rate > -3.0 else "negative")
|
||||||
tags = [strategy, exit_type]
|
tags = [strategy, exit_type]
|
||||||
@@ -819,6 +822,16 @@ async def _handle_insert_trade_journal(session, data: dict) -> None:
|
|||||||
resolved = result.scalar_one_or_none()
|
resolved = result.scalar_one_or_none()
|
||||||
if resolved:
|
if resolved:
|
||||||
payload["trade_id"] = 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:
|
else:
|
||||||
logger.warning(
|
logger.warning(
|
||||||
"⚠️ [Worker] TradeJournal trade_id 미해소 (order_id=%s, %s) — "
|
"⚠️ [Worker] TradeJournal trade_id 미해소 (order_id=%s, %s) — "
|
||||||
|
|||||||
@@ -247,6 +247,85 @@ async def batch_analysis_scheduler():
|
|||||||
raise
|
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():
|
async def post_market_analyzer_scheduler():
|
||||||
"""장 종료 후 16:00 KST 1회 실행 — 매매일지 자동 생성 + 전략 자동 최적화.
|
"""장 종료 후 16:00 KST 1회 실행 — 매매일지 자동 생성 + 전략 자동 최적화.
|
||||||
|
|
||||||
@@ -300,18 +379,9 @@ async def post_market_analyzer_scheduler():
|
|||||||
from sqlalchemy import select, func, and_
|
from sqlalchemy import select, func, and_
|
||||||
|
|
||||||
async with AsyncSessionLocal() as db:
|
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)
|
perf = await analyze_performance(db, days=30)
|
||||||
journal_count = 0
|
journal_count = await _enqueue_journals_v2(uncovered, perf)
|
||||||
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}")
|
|
||||||
if journal_count > 0:
|
if journal_count > 0:
|
||||||
logger.info(f"📝 [PostMarket] 일지 {journal_count}건 큐 등록 완료")
|
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
|
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_win = sum(wins) / len(wins) if wins else 0.0
|
||||||
avg_loss = sum(losses) / len(losses) if losses 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_profit = sum(a for a in pl_amounts if a > 0)
|
||||||
gross_loss = 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
|
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)
|
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:
|
else:
|
||||||
insights.append({"level": "info", "message": f"승률 {win_rate:.1f}% — 양호합니다."})
|
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} — 손실이 수익보다 큽니다. 손절 기준 강화 필요."})
|
insights.append({"level": "danger", "message": f"Profit Factor {pf:.2f} — 손실이 수익보다 큽니다. 손절 기준 강화 필요."})
|
||||||
elif pf < 1.5:
|
elif pf < 1.5:
|
||||||
insights.append({"level": "warning", "message": f"Profit Factor {pf:.2f} — 개선 여지 있습니다."})
|
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]:
|
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)
|
since = now_kst() - timedelta(days=days)
|
||||||
|
cutoff = now_kst() - timedelta(minutes=max(0, int(min_age_minutes)))
|
||||||
|
|
||||||
# 이미 일지가 있는 trade_id 목록
|
# 이미 일지가 있는 trade_id 목록
|
||||||
# [2026-09-29 B3] `> 0` 조건이 필수다. 과거 호출부가 trade_id 미확보 시 `0` 을
|
# [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(
|
stmt = select(TradeHistory).where(
|
||||||
and_(
|
and_(
|
||||||
TradeHistory.order_type == "SELL",
|
TradeHistory.order_type == "SELL",
|
||||||
TradeHistory.status.in_(["COMPLETED", "FILLED"]),
|
TradeHistory.status.in_(JOURNAL_SELL_STATUSES),
|
||||||
TradeHistory.created_at >= since,
|
TradeHistory.created_at >= since,
|
||||||
|
TradeHistory.created_at <= cutoff,
|
||||||
TradeHistory.id.not_in(covered_ids) if covered_ids else True,
|
TradeHistory.id.not_in(covered_ids) if covered_ids else True,
|
||||||
)
|
)
|
||||||
).order_by(TradeHistory.created_at.desc())
|
).order_by(TradeHistory.created_at.desc())
|
||||||
|
|||||||
268
backend/tests/test_journal_intraday_20261001.py
Normal file
268
backend/tests/test_journal_intraday_20261001.py
Normal file
@@ -0,0 +1,268 @@
|
|||||||
|
"""매매일지 회귀 (2026-10-01): 장중 생성 부재 · strategy_name='unknown' · 프로핏팩터.
|
||||||
|
|
||||||
|
- 16:00 배치 enqueue_trade_journal 이 strategy_type 컬럼을 넘기지 않아 전부 'unknown'.
|
||||||
|
- 장중 주기 일지 스케줄러 신설(멱등).
|
||||||
|
- profit_factor 는 손익률(%) 합이 아니라 금액(gross_profit/|gross_loss|) 기준.
|
||||||
|
"""
|
||||||
|
import asyncio
|
||||||
|
import pathlib
|
||||||
|
from types import SimpleNamespace
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
APP_ROOT = pathlib.Path(__file__).resolve().parents[1] / "app"
|
||||||
|
|
||||||
|
|
||||||
|
def _trade(**kw):
|
||||||
|
base = dict(
|
||||||
|
id=1, stock_code="005930", stock_name="삼성전자", order_type="SELL",
|
||||||
|
profit_loss_rate=-2.0, net_profit_loss=-1000, entry_reason="",
|
||||||
|
exit_reason="stop_loss", strategy_type="swing", executed_at=None, created_at=None,
|
||||||
|
executed_price=1, executed_quantity=1, order_quantity=1,
|
||||||
|
)
|
||||||
|
base.update(kw)
|
||||||
|
return SimpleNamespace(**base)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("st,expected", [
|
||||||
|
("swing", "mean_reversion"), ("daytrading", "day_trade"),
|
||||||
|
("day_trade", "day_trade"), ("manual", "manual"),
|
||||||
|
])
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_enqueue_trade_journal_uses_strategy_type(monkeypatch, st, expected):
|
||||||
|
from app.services import db_worker
|
||||||
|
|
||||||
|
sent = []
|
||||||
|
|
||||||
|
async def _put(item):
|
||||||
|
sent.append(item)
|
||||||
|
|
||||||
|
monkeypatch.setattr(db_worker.db_task_queue, "put", _put)
|
||||||
|
await db_worker.enqueue_trade_journal(_trade(strategy_type=st), "c")
|
||||||
|
assert sent[0]["data"]["strategy_name"] == expected
|
||||||
|
assert sent[0]["data"]["tags"][0] == expected
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_batch_and_manual_paths_agree_on_strategy(monkeypatch):
|
||||||
|
from app.services import db_worker
|
||||||
|
from app.services.trade_analyzer import _parse_strategy
|
||||||
|
|
||||||
|
sent = []
|
||||||
|
|
||||||
|
async def _put(item):
|
||||||
|
sent.append(item)
|
||||||
|
|
||||||
|
monkeypatch.setattr(db_worker.db_task_queue, "put", _put)
|
||||||
|
t = _trade(strategy_type="daytrading")
|
||||||
|
await db_worker.enqueue_trade_journal(t, "c")
|
||||||
|
assert sent[0]["data"]["strategy_name"] == _parse_strategy(t.entry_reason, t.strategy_type)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_enqueue_journals_v2_is_idempotent(monkeypatch):
|
||||||
|
from app.services import schedulers, db_worker
|
||||||
|
|
||||||
|
calls = []
|
||||||
|
|
||||||
|
async def _enq(trade, content):
|
||||||
|
calls.append(trade.id)
|
||||||
|
|
||||||
|
monkeypatch.setattr(db_worker, "enqueue_trade_journal", _enq)
|
||||||
|
schedulers._JOURNAL_ENQUEUED_IDS.clear()
|
||||||
|
trades = [_trade(id=11), _trade(id=12)]
|
||||||
|
assert await schedulers._enqueue_journals_v2(trades) == 2
|
||||||
|
assert await schedulers._enqueue_journals_v2(trades) == 0 # 큐 적체 중 재조회
|
||||||
|
assert calls == [11, 12]
|
||||||
|
schedulers._JOURNAL_ENQUEUED_IDS.clear()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_enqueue_failure_is_not_marked_and_retried(monkeypatch):
|
||||||
|
from app.services import schedulers, db_worker
|
||||||
|
|
||||||
|
async def _boom(trade, content):
|
||||||
|
raise RuntimeError("queue down")
|
||||||
|
|
||||||
|
monkeypatch.setattr(db_worker, "enqueue_trade_journal", _boom)
|
||||||
|
schedulers._JOURNAL_ENQUEUED_IDS.clear()
|
||||||
|
assert await schedulers._enqueue_journals_v2([_trade(id=21)]) == 0
|
||||||
|
assert 21 not in schedulers._JOURNAL_ENQUEUED_IDS
|
||||||
|
|
||||||
|
|
||||||
|
def test_intraday_scheduler_wired_and_config_driven():
|
||||||
|
main_src = (APP_ROOT / "main.py").read_text()
|
||||||
|
assert "journal_intraday_scheduler()" in main_src
|
||||||
|
assert '"journal_intraday_task"' in main_src
|
||||||
|
src = (APP_ROOT / "services" / "schedulers.py").read_text()
|
||||||
|
body = src[src.index("async def journal_intraday_scheduler"):src.index("async def post_market_analyzer_scheduler")]
|
||||||
|
assert "settings.JOURNAL_INTRADAY_INTERVAL_MIN" in body
|
||||||
|
assert "commit(" not in body # 실시간 계층 DB 쓰기 금지
|
||||||
|
assert "logger.debug" not in body
|
||||||
|
|
||||||
|
|
||||||
|
def test_uncovered_includes_partial_but_not_pending():
|
||||||
|
from app.services.trade_analyzer import JOURNAL_SELL_STATUSES
|
||||||
|
|
||||||
|
assert "PARTIAL" in JOURNAL_SELL_STATUSES
|
||||||
|
assert "PENDING" not in JOURNAL_SELL_STATUSES and "FAILED" not in JOURNAL_SELL_STATUSES
|
||||||
|
|
||||||
|
|
||||||
|
def test_performance_stats_keep_completed_filter():
|
||||||
|
src = (APP_ROOT / "services" / "trade_analyzer.py").read_text()
|
||||||
|
seg = src[src.index("async def analyze_performance"):src.index("async def analyze_performance") + 2500]
|
||||||
|
assert '["COMPLETED", "FILLED"]' in seg and "PARTIAL" not in seg
|
||||||
|
|
||||||
|
|
||||||
|
class _R:
|
||||||
|
def __init__(self, items):
|
||||||
|
self._i = items
|
||||||
|
|
||||||
|
def scalars(self):
|
||||||
|
return SimpleNamespace(all=lambda: list(self._i))
|
||||||
|
|
||||||
|
|
||||||
|
class _DB:
|
||||||
|
def __init__(self, trades):
|
||||||
|
self.t = trades
|
||||||
|
|
||||||
|
async def execute(self, stmt, *a, **k):
|
||||||
|
return _R(self.t)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_profit_factor_uses_amounts_not_percent_sums():
|
||||||
|
"""실측 재현: gross_profit 266,370 / gross_loss -430,619 인데 PF>1 이던 버그."""
|
||||||
|
from app.services.trade_analyzer import analyze_performance
|
||||||
|
|
||||||
|
trades = [
|
||||||
|
# 소액 대박(+30%, +1,000원) 과 대금 큰 소폭손실(-1%, -430,619원)
|
||||||
|
_trade(id=1, profit_loss_rate=30.0, net_profit_loss=266370),
|
||||||
|
_trade(id=2, profit_loss_rate=-1.0, net_profit_loss=-430619),
|
||||||
|
]
|
||||||
|
perf = await analyze_performance(_DB(trades), days=30)
|
||||||
|
s = perf["summary"]
|
||||||
|
assert s["gross_profit"] == 266370 and s["gross_loss"] == -430619
|
||||||
|
assert s["profit_factor"] == pytest.approx(266370 / 430619, rel=1e-3)
|
||||||
|
assert s["profit_factor"] < 1.0
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_profit_factor_no_loss_capped():
|
||||||
|
from app.services.trade_analyzer import analyze_performance
|
||||||
|
|
||||||
|
perf = await analyze_performance(_DB([_trade(profit_loss_rate=3, net_profit_loss=500)]), days=30)
|
||||||
|
assert perf["summary"]["profit_factor"] == 99.0
|
||||||
|
|
||||||
|
|
||||||
|
# ── 2026-10-01 보완 ──────────────────────────────────────────────────────────
|
||||||
|
class _FR:
|
||||||
|
def __init__(self, items):
|
||||||
|
self._i = list(items)
|
||||||
|
|
||||||
|
def scalar_one_or_none(self):
|
||||||
|
return self._i[0] if self._i else None
|
||||||
|
|
||||||
|
|
||||||
|
class _FSess:
|
||||||
|
def __init__(self, results):
|
||||||
|
self.r = list(results)
|
||||||
|
self.added = []
|
||||||
|
self.commits = 0
|
||||||
|
|
||||||
|
async def execute(self, stmt, *a, **k):
|
||||||
|
return _FR(self.r.pop(0) if self.r else [])
|
||||||
|
|
||||||
|
def add(self, o):
|
||||||
|
self.added.append(o)
|
||||||
|
|
||||||
|
async def commit(self):
|
||||||
|
self.commits += 1
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_handler_skips_when_resolved_trade_already_has_journal():
|
||||||
|
from app.services.db_worker import _handle_insert_trade_journal
|
||||||
|
|
||||||
|
db = _FSess([[334], [77]]) # order_id→334 해소, 기존 일지 id=77 존재
|
||||||
|
await _handle_insert_trade_journal(db, {
|
||||||
|
"trade_id": None, "order_id": "1", "stock_code": "009830", "content": "x",
|
||||||
|
})
|
||||||
|
assert db.added == [] and db.commits == 0
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_handler_explicit_trade_id_still_skips_lookup():
|
||||||
|
from app.services.db_worker import _handle_insert_trade_journal
|
||||||
|
|
||||||
|
db = _FSess([])
|
||||||
|
await _handle_insert_trade_journal(db, {"trade_id": 5, "stock_code": "x", "content": "c"})
|
||||||
|
assert len(db.added) == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_uncovered_v2_has_grace_cutoff_and_intraday_uses_it():
|
||||||
|
import inspect
|
||||||
|
from app.services import trade_analyzer, schedulers
|
||||||
|
|
||||||
|
sig = inspect.signature(trade_analyzer.get_uncovered_trades)
|
||||||
|
assert list(sig.parameters) == ["db", "days"] # 기존 시그니처 불변
|
||||||
|
assert "min_age_minutes" in inspect.signature(trade_analyzer.get_uncovered_trades_v2).parameters
|
||||||
|
body = inspect.getsource(schedulers.journal_intraday_scheduler)
|
||||||
|
assert "get_uncovered_trades_v2" in body and "JOURNAL_INTRADAY_GRACE_MIN" in body
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_uncovered_v2_sql_has_created_at_cutoff():
|
||||||
|
from app.services.trade_analyzer import get_uncovered_trades_v2
|
||||||
|
|
||||||
|
class _D:
|
||||||
|
stmts = []
|
||||||
|
|
||||||
|
async def execute(self, stmt, *a, **k):
|
||||||
|
self.stmts.append(stmt)
|
||||||
|
return SimpleNamespace(fetchall=lambda: [], scalars=lambda: SimpleNamespace(all=lambda: []))
|
||||||
|
|
||||||
|
d = _D()
|
||||||
|
await get_uncovered_trades_v2(d, days=7, min_age_minutes=3)
|
||||||
|
sql = str(d.stmts[1].compile()).replace("\n", " ")
|
||||||
|
assert "trade_history.created_at <=" in sql
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_inflight_entries_expire_after_ttl(monkeypatch):
|
||||||
|
from app.core.config import settings
|
||||||
|
from app.services import schedulers, db_worker
|
||||||
|
|
||||||
|
calls = []
|
||||||
|
|
||||||
|
async def _enq(trade, content):
|
||||||
|
calls.append(trade.id)
|
||||||
|
|
||||||
|
monkeypatch.setattr(db_worker, "enqueue_trade_journal", _enq)
|
||||||
|
schedulers._JOURNAL_ENQUEUED_IDS.clear()
|
||||||
|
t = [_trade(id=31)]
|
||||||
|
assert await schedulers._enqueue_journals_v2(t) == 1
|
||||||
|
assert await schedulers._enqueue_journals_v2(t) == 0
|
||||||
|
# 등록 시각을 TTL 이전으로 되돌리면 재시도된다 (DLQ 로 간 INSERT 복구)
|
||||||
|
schedulers._JOURNAL_ENQUEUED_IDS[31] -= settings.JOURNAL_ENQUEUE_TTL_MIN * 60 + 1
|
||||||
|
assert await schedulers._enqueue_journals_v2(t) == 1
|
||||||
|
assert calls == [31, 31]
|
||||||
|
schedulers._JOURNAL_ENQUEUED_IDS.clear()
|
||||||
|
|
||||||
|
|
||||||
|
def test_journals_endpoint_defenses_and_lookback_config():
|
||||||
|
src = (APP_ROOT / "api" / "endpoints" / "journals.py").read_text()
|
||||||
|
assert "existing.scalars().first()" in src
|
||||||
|
assert "JOURNAL_SELL_STATUSES" in src and '("COMPLETED", "FILLED")' not in src
|
||||||
|
sch = (APP_ROOT / "services" / "schedulers.py").read_text()
|
||||||
|
assert "get_uncovered_trades(db, days=7)" not in sch
|
||||||
|
assert "days=settings.JOURNAL_LOOKBACK_DAYS" in sch
|
||||||
|
|
||||||
|
|
||||||
|
def test_insights_skip_pf_when_no_pnl():
|
||||||
|
from app.services.trade_analyzer import build_insights, _empty_summary
|
||||||
|
|
||||||
|
out = build_insights({"summary": _empty_summary()})
|
||||||
|
assert not any("Profit Factor" in i["message"] for i in out)
|
||||||
|
out2 = build_insights({"summary": {**_empty_summary(), "gross_loss": -100.0, "profit_factor": 0.5}})
|
||||||
|
assert any("Profit Factor" in i["message"] for i in out2)
|
||||||
Reference in New Issue
Block a user