feat: 매매일지 자동 생성 + 전략 자동 최적화 시스템 추가

- trade_analyzer.py: 거래 성과 분석(승률/손익비/전략별 패턴), 템플릿 기반 일지 자동 생성
- strategy_optimizer.py: 그리드 서치 기반 파라미터 최적화, Sharpe 10% 개선 시 자동 적용
- analysis.py: /performance /insights /generate-journals /optimize-strategy /optimization-history 엔드포인트
- journals.py: POST /generate/{trade_id}, GET /stats 엔드포인트 추가
- models.py: StrategyOptimizationLog 테이블 추가
- main.py: post_market_analyzer_scheduler (16:00 KST 자동 실행)
- frontend: Analysis 페이지 추가 (성과 요약/인사이트/일지 생성/최적화 이력)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-10 09:28:10 +09:00
parent f42e94fe6c
commit 2baf06a5fd
10 changed files with 1789 additions and 5 deletions

View File

@@ -7,7 +7,7 @@ from fastapi import APIRouter
from app.api.endpoints import ( from app.api.endpoints import (
trading_rules, trades, portfolio, market_data, mcp, recommendations, trading_rules, trades, portfolio, market_data, mcp, recommendations,
monitoring, tensorflow_predictions, debug, risk_management, ml_ensemble, monitoring, tensorflow_predictions, debug, risk_management, ml_ensemble,
journals, day_trading, strategy_config, backtest, journals, day_trading, strategy_config, backtest, analysis,
) )
api_router = APIRouter() api_router = APIRouter()
@@ -27,3 +27,4 @@ api_router.include_router(journals.router, prefix="/journals", tags=["journals"]
api_router.include_router(day_trading.router, prefix="/daytrading", tags=["day_trading"]) api_router.include_router(day_trading.router, prefix="/daytrading", tags=["day_trading"])
api_router.include_router(strategy_config.router, prefix="/strategy-config", tags=["strategy_config"]) api_router.include_router(strategy_config.router, prefix="/strategy-config", tags=["strategy_config"])
api_router.include_router(backtest.router, prefix="/backtest", tags=["backtest"]) api_router.include_router(backtest.router, prefix="/backtest", tags=["backtest"])
api_router.include_router(analysis.router, prefix="/analysis", tags=["analysis"])

View File

@@ -0,0 +1,242 @@
"""
거래 성과 분석 + 전략 자동 최적화 API
GET /api/v1/analysis/performance?days=30 — 성과 통계
GET /api/v1/analysis/insights?days=30 — 전략 인사이트 + 개선 제안
POST /api/v1/analysis/generate-journals — 미생성 일지 일괄 생성
POST /api/v1/analysis/optimize-strategy — 전략 최적화 즉시 실행
GET /api/v1/analysis/optimization-history — 최적화 이력
"""
import logging
from datetime import datetime, date
from typing import Any, Dict, List, Optional
from fastapi import APIRouter, BackgroundTasks, Depends, Query
from sqlalchemy import select, desc
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db, AsyncSessionLocal
from app.models.models import StrategyOptimizationLog, TradeJournal
from app.services.trade_analyzer import (
analyze_performance,
build_insights,
create_journal_record,
generate_journal_for_trade,
get_uncovered_trades,
)
from app.services.strategy_optimizer import (
get_last_optimization_date,
run_optimization,
save_optimization_log,
)
router = APIRouter()
logger = logging.getLogger(__name__)
# ─────────────────────────────────────────────────────────────────────────────
# GET /performance
# ─────────────────────────────────────────────────────────────────────────────
@router.get("/performance")
async def get_performance(
days: int = Query(default=30, ge=1, le=365, description="분석 기간 (일)"),
db: AsyncSession = Depends(get_db),
) -> Dict[str, Any]:
"""최근 N일 완료 거래 성과 통계 반환."""
try:
return await analyze_performance(db, days=days)
except Exception as e:
logger.error(f"performance 분석 실패: {e}", exc_info=True)
return {"error": str(e), "summary": {}, "total_trades": 0}
# ─────────────────────────────────────────────────────────────────────────────
# GET /insights
# ─────────────────────────────────────────────────────────────────────────────
@router.get("/insights")
async def get_insights(
days: int = Query(default=30, ge=1, le=365),
db: AsyncSession = Depends(get_db),
) -> Dict[str, Any]:
"""전략 인사이트 및 개선 제안 반환."""
try:
perf = await analyze_performance(db, days=days)
insights = build_insights(perf)
last_opt_date = await get_last_optimization_date(db)
return {
"insights": insights,
"performance_summary": perf.get("summary", {}),
"last_optimization_date": last_opt_date.isoformat() if last_opt_date else None,
"period_days": days,
"generated_at": datetime.utcnow().isoformat(),
}
except Exception as e:
logger.error(f"insights 생성 실패: {e}", exc_info=True)
return {"insights": [], "error": str(e)}
# ─────────────────────────────────────────────────────────────────────────────
# POST /generate-journals
# ─────────────────────────────────────────────────────────────────────────────
async def _generate_journals_task(days: int = 7) -> Dict[str, Any]:
"""백그라운드에서 미생성 일지를 일괄 생성."""
async with AsyncSessionLocal() as db:
try:
trades = await get_uncovered_trades(db, days=days)
if not trades:
return {"created": 0, "message": "생성할 미처리 거래 없음"}
perf = await analyze_performance(db, days=30)
created = 0
for trade in trades:
try:
content = generate_journal_for_trade(trade, perf)
await create_journal_record(db, trade, content)
created += 1
except Exception as e:
logger.warning(f"일지 생성 실패 ({trade.stock_code}): {e}")
await db.commit()
logger.info(f"✅ 매매일지 {created}건 생성 완료 (대상: {len(trades)}건)")
return {"created": created, "total_uncovered": len(trades)}
except Exception as e:
await db.rollback()
logger.error(f"일지 일괄 생성 실패: {e}", exc_info=True)
return {"created": 0, "error": str(e)}
@router.post("/generate-journals")
async def generate_journals(
background_tasks: BackgroundTasks,
days: int = Query(default=7, ge=1, le=30, description="최근 N일 내 미생성 거래 대상"),
db: AsyncSession = Depends(get_db),
) -> Dict[str, Any]:
"""일지가 없는 완료 거래에 대해 템플릿 기반 매매일지를 일괄 생성."""
try:
uncovered = await get_uncovered_trades(db, days=days)
count = len(uncovered)
if count == 0:
return {"status": "skipped", "message": "생성할 미처리 거래 없음", "queued": 0}
background_tasks.add_task(_generate_journals_task, days)
return {
"status": "started",
"message": f"{count}건의 거래 일지 생성을 시작했습니다.",
"queued": count,
}
except Exception as e:
logger.error(f"generate-journals 실패: {e}", exc_info=True)
return {"status": "error", "message": str(e)}
# ─────────────────────────────────────────────────────────────────────────────
# POST /optimize-strategy
# ─────────────────────────────────────────────────────────────────────────────
# 최적화 진행 상태 (단순 인메모리 플래그)
_optimization_running: bool = False
@router.post("/optimize-strategy")
async def optimize_strategy(
background_tasks: BackgroundTasks,
days: int = Query(default=30, ge=7, le=180, description="성과 분석 기간"),
period_days: int = Query(default=90, ge=30, le=365, description="백테스트 기간"),
) -> Dict[str, Any]:
"""
전략 파라미터 최적화를 즉시 실행 (수동 트리거).
- 성과 분석 → 파라미터 그리드 서치 → Sharpe 10% 개선 시 자동 적용
- 백그라운드에서 실행 (수 분 소요 가능)
"""
global _optimization_running
if _optimization_running:
return {"status": "already_running", "message": "최적화가 이미 실행 중입니다."}
async def _opt_task():
global _optimization_running
_optimization_running = True
try:
async with AsyncSessionLocal() as db:
perf = await analyze_performance(db, days=days)
result = await run_optimization(db, perf, trigger="manual", period_days=period_days)
await save_optimization_log(db, result, trigger="manual")
logger.info(f"✅ 수동 최적화 완료: applied={result.applied}, reason={result.apply_reason}")
except Exception as e:
logger.error(f"최적화 실패: {e}", exc_info=True)
finally:
_optimization_running = False
background_tasks.add_task(_opt_task)
return {
"status": "started",
"message": "전략 최적화를 시작했습니다. /optimization-history 에서 결과를 확인하세요.",
"performance_days": days,
"backtest_period_days": period_days,
}
# ─────────────────────────────────────────────────────────────────────────────
# GET /optimization-history
# ─────────────────────────────────────────────────────────────────────────────
@router.get("/optimization-history")
async def get_optimization_history(
limit: int = Query(default=10, ge=1, le=50),
db: AsyncSession = Depends(get_db),
) -> List[Dict[str, Any]]:
"""전략 최적화 실행 이력 반환."""
try:
stmt = (
select(StrategyOptimizationLog)
.order_by(desc(StrategyOptimizationLog.created_at))
.limit(limit)
)
result = await db.execute(stmt)
logs = result.scalars().all()
return [
{
"id": log.id,
"run_date": log.run_date.isoformat() if log.run_date else None,
"trigger": log.trigger,
"period_days": log.period_days,
"best_sharpe": float(log.best_sharpe) if log.best_sharpe else None,
"current_sharpe": float(log.current_sharpe) if log.current_sharpe else None,
"sharpe_improvement_pct": (
round(
(float(log.best_sharpe) - float(log.current_sharpe))
/ max(abs(float(log.current_sharpe)), 0.001)
* 100,
2,
)
if log.best_sharpe and log.current_sharpe
else None
),
"best_win_rate": float(log.best_win_rate) if log.best_win_rate else None,
"best_mdd": float(log.best_mdd) if log.best_mdd else None,
"applied": log.applied,
"apply_reason": log.apply_reason,
"best_params": log.best_params,
"previous_params": log.previous_params,
"total_combinations": log.total_combinations,
"created_at": log.created_at.isoformat() if log.created_at else None,
}
for log in logs
]
except Exception as e:
logger.error(f"optimization-history 조회 실패: {e}", exc_info=True)
return []
# ─────────────────────────────────────────────────────────────────────────────
# GET /optimization-status
# ─────────────────────────────────────────────────────────────────────────────
@router.get("/optimization-status")
async def get_optimization_status() -> Dict[str, Any]:
"""현재 최적화 실행 상태 조회."""
return {"running": _optimization_running}

View File

@@ -2,11 +2,13 @@
매매일지 API 엔드포인트 매매일지 API 엔드포인트
- GET /: 최근 AI 생성 매매일지 목록 - GET /: 최근 AI 생성 매매일지 목록
- GET /stats: 일지 통계 (전략별, 감정 분포)
- GET /{journal_id}: 특정 일지 상세 - GET /{journal_id}: 특정 일지 상세
- POST /generate/{trade_id}: 특정 거래에 대한 일지 즉시 생성
""" """
from fastapi import APIRouter, Depends, Query from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, desc from sqlalchemy import select, desc, func
from typing import Optional from typing import Optional
from datetime import datetime, timezone from datetime import datetime, timezone
import logging import logging
@@ -19,7 +21,7 @@ def _utc_iso(dt: datetime | None) -> str | None:
return dt.replace(tzinfo=timezone.utc).isoformat() return dt.replace(tzinfo=timezone.utc).isoformat()
from app.core.database import get_db from app.core.database import get_db
from app.models.models import TradeJournal from app.models.models import TradeJournal, TradeHistory
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -63,6 +65,81 @@ async def list_journals(
return [] return []
@router.get("/stats")
async def get_journal_stats(db: AsyncSession = Depends(get_db)):
"""일지 통계: 전략별 / 감정별 분포."""
try:
result_mood = await db.execute(
select(TradeJournal.mood, func.count(TradeJournal.id).label("cnt"))
.group_by(TradeJournal.mood)
)
mood_dist = {row[0] or "unknown": row[1] for row in result_mood.fetchall()}
result_strat = await db.execute(
select(TradeJournal.strategy_name, func.count(TradeJournal.id).label("cnt"))
.group_by(TradeJournal.strategy_name)
)
strat_dist = {row[0] or "unknown": row[1] for row in result_strat.fetchall()}
total = sum(mood_dist.values())
return {
"total": total,
"by_mood": mood_dist,
"by_strategy": strat_dist,
}
except Exception as e:
logger.error(f"journal stats error: {e}", exc_info=True)
return {"total": 0, "by_mood": {}, "by_strategy": {}}
@router.post("/generate/{trade_id}")
async def generate_journal_for_trade_id(
trade_id: int,
db: AsyncSession = Depends(get_db),
):
"""특정 완료 거래(trade_id)에 대해 템플릿 기반 매매일지를 즉시 생성."""
from app.services.trade_analyzer import (
analyze_performance,
generate_journal_for_trade,
create_journal_record,
)
# 기존 일지 확인
existing = await db.execute(
select(TradeJournal).where(TradeJournal.trade_id == trade_id)
)
if existing.scalar_one_or_none():
raise HTTPException(status_code=409, detail="이미 일지가 존재합니다.")
# 거래 조회
trade_result = await db.execute(
select(TradeHistory).where(TradeHistory.id == trade_id)
)
trade = trade_result.scalar_one_or_none()
if not trade:
raise HTTPException(status_code=404, detail="거래 이력을 찾을 수 없습니다.")
if trade.order_type != "SELL" or trade.status != "FILLED":
raise HTTPException(status_code=400, detail="완료된 매도 거래에 대해서만 일지를 생성할 수 있습니다.")
perf = await analyze_performance(db, days=30)
content = generate_journal_for_trade(trade, perf)
journal = await create_journal_record(db, trade, content)
await db.commit()
await db.refresh(journal)
return {
"id": journal.id,
"trade_id": journal.trade_id,
"stock_code": journal.stock_code,
"stock_name": journal.stock_name,
"journal_date": _utc_iso(journal.journal_date),
"content": journal.content,
"strategy_name": journal.strategy_name,
"mood": journal.mood,
"tags": journal.tags,
"created_at": _utc_iso(journal.created_at),
}
@router.get("/{journal_id}") @router.get("/{journal_id}")
async def get_journal(journal_id: int, db: AsyncSession = Depends(get_db)): async def get_journal(journal_id: int, db: AsyncSession = Depends(get_db)):
"""특정 매매일지 상세 조회""" """특정 매매일지 상세 조회"""

View File

@@ -372,6 +372,111 @@ async def day_trade_scheduler():
raise raise
async def post_market_analyzer_scheduler():
"""장 종료 후 16:00 KST 1회 실행 — 매매일지 자동 생성 + 전략 자동 최적화.
조건:
- 매일 16:00 KST 에 1회 실행 (이미 오늘 실행했으면 스킵)
- 최근 7일 내 SELL FILLED 거래 >= 5건이면 최적화 실행
- 마지막 최적화로부터 3일 이상 경과 시만 최적화 실행
"""
import pytz as _pytz
from datetime import datetime as _dt, date as _date, timedelta as _td
_KST = _pytz.timezone("Asia/Seoul")
_last_run: _date | None = None
await asyncio.sleep(60) # 부팅 안정화
logger.info("📝 Post-market analyzer scheduler started (daily 16:00 KST)")
try:
while True:
now_kst = _dt.now(_KST)
today = now_kst.date()
if _last_run == today:
tomorrow_16 = _KST.localize(_dt(today.year, today.month, today.day, 16, 0, 0)) + _td(days=1)
sleep_sec = (tomorrow_16 - now_kst).total_seconds()
await asyncio.sleep(max(sleep_sec, 60))
continue
if (now_kst.hour, now_kst.minute) < (16, 0):
target = _KST.localize(_dt(today.year, today.month, today.day, 16, 0, 0))
sleep_sec = (target - now_kst).total_seconds()
if sleep_sec > 0:
await asyncio.sleep(min(sleep_sec, 1800))
continue
if today.weekday() >= 5:
await asyncio.sleep(3600)
continue
logger.info("📝 [PostMarket] 매매일지 자동 생성 + 전략 최적화 시작")
_last_run = today
try:
from app.services.trade_analyzer import (
analyze_performance,
get_uncovered_trades,
generate_journal_for_trade,
create_journal_record,
)
from app.services.strategy_optimizer import (
run_optimization,
save_optimization_log,
get_last_optimization_date,
)
from app.models.models import TradeHistory
from sqlalchemy import func, and_
from datetime import datetime as _utcnow_dt
async with AsyncSessionLocal() as db:
uncovered = await get_uncovered_trades(db, days=7)
perf = await analyze_performance(db, days=30)
journal_count = 0
for trade in uncovered:
try:
content = generate_journal_for_trade(trade, perf)
await create_journal_record(db, trade, content)
journal_count += 1
except Exception as e:
logger.warning(f"일지 생성 실패 ({trade.stock_code}): {e}")
if journal_count > 0:
await db.commit()
logger.info(f"📝 [PostMarket] 일지 {journal_count}건 생성 완료")
since_7d = _utcnow_dt.utcnow() - _td(days=7)
trade_count_result = await db.execute(
select(func.count(TradeHistory.id)).where(
and_(
TradeHistory.order_type == "SELL",
TradeHistory.status == "FILLED",
TradeHistory.created_at >= since_7d,
)
)
)
recent_trade_count = trade_count_result.scalar() or 0
last_opt = await get_last_optimization_date(db)
days_since_opt = (today - last_opt).days if last_opt else 999
if recent_trade_count >= 5 and days_since_opt >= 3:
logger.info(f"🔧 [PostMarket] 전략 최적화 실행 (최근거래:{recent_trade_count}건, 마지막:{days_since_opt}일 전)")
result = await run_optimization(db, perf, trigger="auto", period_days=90)
await save_optimization_log(db, result, trigger="auto")
logger.info(f"✅ [PostMarket] 최적화 완료: applied={result.applied}, {result.apply_reason}")
else:
logger.info(f"⏭️ [PostMarket] 최적화 스킵 (거래:{recent_trade_count}건, 마지막최적화:{days_since_opt}일 전)")
except Exception as e:
logger.error(f"❌ [PostMarket] 분석 실패: {e}", exc_info=True)
await asyncio.sleep(3600)
except asyncio.CancelledError:
logger.info("⏹️ Post-market analyzer scheduler stopped")
raise
async def krx_data_collector_scheduler(): async def krx_data_collector_scheduler():
"""장 종료 후 KRX 일봉 데이터 자동 수집 스케줄러. """장 종료 후 KRX 일봉 데이터 자동 수집 스케줄러.
@@ -550,6 +655,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.day_trade_task = asyncio.create_task(day_trade_scheduler()) app.state.day_trade_task = asyncio.create_task(day_trade_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())
# KIS 실시간 스트림 시작 (설정으로 제어 가능) # KIS 실시간 스트림 시작 (설정으로 제어 가능)
if settings.KIS_WS_ENABLED: if settings.KIS_WS_ENABLED:
@@ -568,6 +674,7 @@ async def lifespan(app: FastAPI):
"batch_analysis_task", "batch_analysis_task",
"day_trade_task", "day_trade_task",
"krx_collector_task", "krx_collector_task",
"post_market_task",
"db_worker_task", "db_worker_task",
"kis_stream_task", "kis_stream_task",
): ):

View File

@@ -322,3 +322,23 @@ class WatchlistHistory(Base):
change_rate = Column(Numeric(10, 4), nullable=True) change_rate = Column(Numeric(10, 4), nullable=True)
rsi = Column(Numeric(10, 4), nullable=True) rsi = Column(Numeric(10, 4), nullable=True)
signals = Column(JSONB, nullable=True) signals = Column(JSONB, nullable=True)
class StrategyOptimizationLog(Base):
"""전략 자동 최적화 실행 이력"""
__tablename__ = "strategy_optimization_logs"
id = Column(Integer, primary_key=True, index=True)
run_date = Column(Date, nullable=False, index=True)
trigger = Column(String(20), nullable=False, default="auto") # auto / manual
period_days = Column(Integer, nullable=False, default=90)
best_params = Column(JSONB, nullable=True)
best_sharpe = Column(Numeric(10, 4), nullable=True)
best_win_rate = Column(Numeric(10, 4), nullable=True)
best_mdd = Column(Numeric(10, 4), nullable=True)
current_sharpe = Column(Numeric(10, 4), nullable=True)
previous_params = Column(JSONB, nullable=True)
applied = Column(Boolean, default=False, nullable=False)
apply_reason = Column(Text, nullable=True)
total_combinations = Column(Integer, nullable=True)
created_at = Column(DateTime, default=datetime.utcnow, index=True)

View File

@@ -0,0 +1,407 @@
"""
전략 자동 최적화 서비스
[역할]
- 최근 거래 성과 약점 분석 → 파라미터 범위 결정
- BacktestEngine을 사용한 그리드 서치로 최적 파라미터 탐색
- 평가 지표: Sharpe * 0.4 + win_rate * 0.3 + (1 - |MDD|/100) * 0.3
- Sharpe 10% 이상 개선 시 strategy_config 자동 적용
- StrategyOptimizationLog 테이블에 이력 저장
[사용법]
from app.services.strategy_optimizer import run_optimization
async with AsyncSessionLocal() as db:
result = await run_optimization(db, perf_data, trigger="manual")
"""
import asyncio
import logging
from datetime import date, datetime, timedelta
from typing import Any, Dict, List, Optional, Tuple
from dataclasses import dataclass
from sqlalchemy import select, desc
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import AsyncSessionLocal
from app.models.models import StrategyOptimizationLog, StockDailyPrice
from app.services.strategy_config_service import strategy_cfg
from app.services.backtest_engine import BacktestEngine, BacktestConfig
logger = logging.getLogger(__name__)
# ─────────────────────────────────────────────────────────────────────────────
# 결과 데이터클래스
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class OptimizationResult:
best_params: Dict[str, Any]
best_score: float
best_sharpe: float
best_win_rate: float
best_mdd: float
current_sharpe: float
previous_params: Dict[str, Any]
applied: bool
apply_reason: str
total_combinations: int
# ─────────────────────────────────────────────────────────────────────────────
# 파라미터 범위 결정 (성과 약점 기반)
# ─────────────────────────────────────────────────────────────────────────────
def _build_param_grid(perf_data: Dict[str, Any]) -> Dict[str, List[Any]]:
"""
성과 약점을 분석하여 그리드 서치할 파라미터 범위를 결정.
규칙:
- 승률 < 50% → RSI_OVERSOLD_LEVEL 강화 (더 낮은 값)
- avg_loss > 손절 기준 → ATR_STOP_MULTIPLIER, STOP_LOSS_PERCENT 조정
- MDD 과다(< -15%) → MAX_POSITION_SIZE, MAX_HOLDINGS_COUNT 조정
- 항상 RSI 기간 변화도 포함 (성과 개선 여지 높음)
"""
summary = perf_data.get("summary", {})
win_rate = summary.get("win_rate", 50.0)
avg_loss = summary.get("avg_loss_pct", -5.0)
grid: Dict[str, List[Any]] = {}
# RSI_OVERSOLD_LEVEL: 승률 기반
if win_rate < 40:
grid["RSI_OVERSOLD_LEVEL"] = [20.0, 25.0, 28.0]
elif win_rate < 50:
grid["RSI_OVERSOLD_LEVEL"] = [25.0, 28.0, 30.0]
else:
grid["RSI_OVERSOLD_LEVEL"] = [28.0, 30.0, 32.0]
# STOP_LOSS_PERCENT: 평균 손실 기반
if avg_loss < -8.0:
grid["STOP_LOSS_PERCENT"] = [-5.0, -6.0, -7.0]
elif avg_loss < -6.0:
grid["STOP_LOSS_PERCENT"] = [-5.0, -6.0]
else:
grid["STOP_LOSS_PERCENT"] = [-6.0, -7.0, -8.0]
# ATR_STOP_MULTIPLIER: 항상 포함
grid["ATR_STOP_MULTIPLIER"] = [2.0, 2.5, 3.0]
# BOLLINGER_STDDEV
grid["BOLLINGER_STDDEV"] = [1.8, 2.0, 2.2]
return grid
def _cartesian_product(grid: Dict[str, List[Any]]) -> List[Dict[str, Any]]:
"""딕셔너리 그리드의 카르테시안 곱을 반환."""
keys = list(grid.keys())
values = list(grid.values())
def _product(lists):
result = [[]]
for lst in lists:
result = [x + [y] for x in result for y in lst]
return result
combos = _product(values)
return [dict(zip(keys, combo)) for combo in combos]
# ─────────────────────────────────────────────────────────────────────────────
# 현재 전략 파라미터 수집
# ─────────────────────────────────────────────────────────────────────────────
async def _get_current_params() -> Dict[str, Any]:
"""현재 strategy_cfg 값들을 읽어 반환."""
await strategy_cfg._refresh()
return {
"RSI_OVERSOLD_LEVEL": strategy_cfg.get("RSI_OVERSOLD_LEVEL", 30.0),
"STOP_LOSS_PERCENT": strategy_cfg.get("STOP_LOSS_PERCENT", -6.0),
"ATR_STOP_MULTIPLIER": strategy_cfg.get("ATR_STOP_MULTIPLIER", 2.5),
"BOLLINGER_STDDEV": strategy_cfg.get("BOLLINGER_STDDEV", 2.0),
"MAX_POSITION_SIZE": strategy_cfg.get("MAX_POSITION_SIZE", 0.04),
"MAX_HOLDINGS_COUNT": strategy_cfg.get("MAX_HOLDINGS_COUNT", 20),
}
# ─────────────────────────────────────────────────────────────────────────────
# 백테스트로 파라미터 평가
# ─────────────────────────────────────────────────────────────────────────────
def _compute_score(sharpe: float, win_rate: float, mdd: float) -> float:
"""통합 평가 지표 계산: Sharpe*0.4 + win_rate*0.3 + (1-|mdd|/100)*0.3"""
mdd_penalty = max(0.0, 1.0 - abs(mdd) / 100.0)
return sharpe * 0.4 + (win_rate / 100.0) * 0.3 + mdd_penalty * 0.3
async def _evaluate_params(
params: Dict[str, Any],
stock_codes: List[str],
start_date: date,
end_date: date,
initial_capital: float,
) -> Tuple[float, float, float, float]:
"""
단일 파라미터 조합으로 백테스트 실행.
Returns: (score, sharpe, win_rate, mdd)
"""
try:
backtest_params = {
"MEAN_REVERSION_RSI_THRESHOLD": params.get("RSI_OVERSOLD_LEVEL", 30.0),
"MEAN_REVERSION_STOP_LOSS_PCT": params.get("STOP_LOSS_PERCENT", -6.0),
"MAX_TOTAL_POSITIONS": min(5, len(stock_codes)),
}
config = BacktestConfig(
stock_codes=stock_codes,
start_date=start_date,
end_date=end_date,
initial_capital=initial_capital,
strategy_params=backtest_params,
max_positions=min(5, len(stock_codes)),
position_size=params.get("MAX_POSITION_SIZE", 0.2),
)
engine = BacktestEngine()
result = await engine.run_backtest(config)
sharpe = float(result.sharpe_ratio or 0.0)
win_rate = float(result.win_rate or 0.0)
mdd = float(result.mdd_pct or 0.0)
score = _compute_score(sharpe, win_rate, mdd)
return score, sharpe, win_rate, mdd
except Exception as e:
logger.debug(f"백테스트 실패 (파라미터: {params}): {e}")
return -1.0, 0.0, 0.0, 0.0
# ─────────────────────────────────────────────────────────────────────────────
# 대표 종목 선택
# ─────────────────────────────────────────────────────────────────────────────
async def _get_representative_stocks(period_days: int = 90) -> List[str]:
"""
최근 N일 거래가 있는 대표 종목 10개 반환.
백테스트 실행 시간을 줄이기 위해 종목 수 제한.
"""
from app.models.models import TradeHistory
from sqlalchemy import func, and_
from datetime import datetime, timedelta
since = datetime.utcnow() - timedelta(days=period_days)
async with AsyncSessionLocal() as db:
stmt = (
select(TradeHistory.stock_code, func.count(TradeHistory.id).label("cnt"))
.where(TradeHistory.created_at >= since)
.group_by(TradeHistory.stock_code)
.order_by(desc("cnt"))
.limit(10)
)
result = await db.execute(stmt)
codes = [row[0] for row in result.fetchall()]
# 거래 이력 없으면 stock_daily_prices에서 최근 종목 fallback
if not codes:
async with AsyncSessionLocal() as db:
stmt = (
select(StockDailyPrice.stock_code)
.where(StockDailyPrice.trade_date >= (date.today() - timedelta(days=period_days)))
.group_by(StockDailyPrice.stock_code)
.limit(10)
)
result = await db.execute(stmt)
codes = [row[0] for row in result.fetchall()]
return codes or []
# ─────────────────────────────────────────────────────────────────────────────
# 파라미터 적용
# ─────────────────────────────────────────────────────────────────────────────
async def apply_params(params: Dict[str, Any], reason: str = "auto_optimizer") -> bool:
"""strategy_cfg에 파라미터를 일괄 적용."""
try:
for key, value in params.items():
await strategy_cfg.upsert(key, str(value), updated_by=reason)
logger.info(f"✅ 전략 파라미터 자동 적용 완료: {list(params.keys())} (사유: {reason})")
return True
except Exception as e:
logger.error(f"❌ 파라미터 적용 실패: {e}", exc_info=True)
return False
# ─────────────────────────────────────────────────────────────────────────────
# 최적화 이력 저장
# ─────────────────────────────────────────────────────────────────────────────
async def save_optimization_log(db: AsyncSession, result: OptimizationResult, trigger: str) -> StrategyOptimizationLog:
"""최적화 결과를 DB에 저장."""
log = StrategyOptimizationLog(
run_date=date.today(),
trigger=trigger,
period_days=90,
best_params=result.best_params,
best_sharpe=result.best_sharpe,
best_win_rate=result.best_win_rate,
best_mdd=result.best_mdd,
current_sharpe=result.current_sharpe,
previous_params=result.previous_params,
applied=result.applied,
apply_reason=result.apply_reason,
total_combinations=result.total_combinations,
)
db.add(log)
await db.commit()
await db.refresh(log)
return log
# ─────────────────────────────────────────────────────────────────────────────
# 메인 최적화 함수
# ─────────────────────────────────────────────────────────────────────────────
async def run_optimization(
db: AsyncSession,
perf_data: Dict[str, Any],
trigger: str = "auto",
period_days: int = 90,
initial_capital: float = 10_000_000,
) -> OptimizationResult:
"""
성과 데이터 기반 전략 파라미터 최적화.
Args:
db: AsyncSession
perf_data: analyze_performance() 반환값
trigger: "auto" | "manual"
period_days: 백테스트 기간 (일)
initial_capital: 백테스트 초기 자본
Returns:
OptimizationResult
"""
logger.info(f"🔧 전략 최적화 시작 (trigger={trigger}, period={period_days}일)")
# 1. 현재 파라미터 저장 (롤백용)
prev_params = await _get_current_params()
# 2. 그리드 구성
param_grid = _build_param_grid(perf_data)
combinations = _cartesian_product(param_grid)
total_combos = len(combinations)
logger.info(f" 그리드 조합 수: {total_combos}")
# 3. 대표 종목 선택
stock_codes = await _get_representative_stocks(period_days)
if not stock_codes:
logger.warning("백테스트 대상 종목 없음. 최적화 건너뜀.")
return OptimizationResult(
best_params=prev_params,
best_score=0.0,
best_sharpe=0.0,
best_win_rate=0.0,
best_mdd=0.0,
current_sharpe=0.0,
previous_params=prev_params,
applied=False,
apply_reason="no_stock_data",
total_combinations=0,
)
end_date = date.today()
start_date = end_date - timedelta(days=period_days)
# 4. 현재 파라미터 기준 Sharpe 계산
cur_score, cur_sharpe, cur_wr, cur_mdd = await _evaluate_params(
prev_params, stock_codes, start_date, end_date, initial_capital
)
logger.info(f" 현재 성과 → Sharpe: {cur_sharpe:.3f}, 승률: {cur_wr:.1f}%, MDD: {cur_mdd:.1f}%")
# 5. 그리드 서치 (배치 5개씩 병렬)
best_score = cur_score
best_params = dict(prev_params)
best_sharpe = cur_sharpe
best_win_rate = cur_wr
best_mdd = cur_mdd
batch_size = 5
for i in range(0, len(combinations), batch_size):
batch = combinations[i:i + batch_size]
tasks = [
_evaluate_params(p, stock_codes, start_date, end_date, initial_capital)
for p in batch
]
results = await asyncio.gather(*tasks, return_exceptions=True)
for combo, res in zip(batch, results):
if isinstance(res, Exception):
continue
score, sharpe, wr, mdd = res
if score > best_score:
best_score = score
best_sharpe = sharpe
best_win_rate = wr
best_mdd = mdd
best_params = {**prev_params, **combo}
logger.info(f" 최적 성과 → Sharpe: {best_sharpe:.3f}, 승률: {best_win_rate:.1f}%, MDD: {best_mdd:.1f}%")
# 6. 자동 적용 여부 결정 (Sharpe 10% 이상 개선)
improvement_threshold = 1.1
should_apply = (
cur_sharpe > 0
and best_sharpe >= cur_sharpe * improvement_threshold
and best_params != prev_params
)
if cur_sharpe <= 0 and best_sharpe > 0:
should_apply = True
applied = False
apply_reason = ""
if should_apply:
improvement_pct = ((best_sharpe - cur_sharpe) / max(abs(cur_sharpe), 0.001)) * 100
apply_reason = f"Sharpe {improvement_pct:.1f}% 향상 ({cur_sharpe:.3f} → {best_sharpe:.3f})"
# 최적화된 파라미터만 적용 (prev와 다른 키만)
changed = {k: v for k, v in best_params.items() if str(v) != str(prev_params.get(k, ""))}
if changed:
applied = await apply_params(changed, reason="auto_optimizer")
if applied:
logger.info(f"✅ 자동 적용 완료: {apply_reason}")
else:
if cur_sharpe <= 0 and best_sharpe <= 0:
apply_reason = "insufficient_data"
elif best_sharpe < cur_sharpe * improvement_threshold:
pct = ((best_sharpe - cur_sharpe) / max(abs(cur_sharpe), 0.001)) * 100
apply_reason = f"개선 미달 ({pct:+.1f}% < 10%)"
else:
apply_reason = "no_improvement"
logger.info(f"⏭️ 자동 적용 건너뜀: {apply_reason}")
return OptimizationResult(
best_params=best_params,
best_score=best_score,
best_sharpe=best_sharpe,
best_win_rate=best_win_rate,
best_mdd=best_mdd,
current_sharpe=cur_sharpe,
previous_params=prev_params,
applied=applied,
apply_reason=apply_reason,
total_combinations=total_combos,
)
# ─────────────────────────────────────────────────────────────────────────────
# 마지막 최적화 이력 조회
# ─────────────────────────────────────────────────────────────────────────────
async def get_last_optimization_date(db: AsyncSession) -> Optional[date]:
"""마지막 최적화 실행일 반환 (없으면 None)."""
stmt = select(StrategyOptimizationLog.run_date).order_by(
desc(StrategyOptimizationLog.created_at)
).limit(1)
result = await db.execute(stmt)
row = result.scalar_one_or_none()
return row

View File

@@ -0,0 +1,438 @@
"""
거래 성과 분석 서비스
[역할]
- TradeHistory 레코드를 분석해 전략 성과 통계를 생성
- 전략별 / 요일별 승률, profit factor, 패턴 분석
- 완료 거래 대상 매매일지 템플릿 자동 생성
- journals.py / analysis.py 엔드포인트에서 사용
"""
import logging
from datetime import datetime, timedelta, date
from typing import Any, Dict, List, Optional
from decimal import Decimal
from sqlalchemy import select, and_, func
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.models import TradeHistory, TradeJournal
logger = logging.getLogger(__name__)
# ─────────────────────────────────────────────────────────────────────────────
# 내부 헬퍼
# ─────────────────────────────────────────────────────────────────────────────
def _to_float(val) -> float:
if val is None:
return 0.0
try:
return float(val)
except (TypeError, ValueError):
return 0.0
def _parse_strategy(entry_reason: Optional[str]) -> str:
"""entry_reason 문자열에서 전략 이름 추출."""
if not entry_reason:
return "unknown"
reason = entry_reason.lower()
if "mean_reversion" in reason or "평균회귀" in reason:
return "mean_reversion"
if "short_term" in reason or "단기" in reason:
return "short_term"
if "kosdaq" in reason or "코스닥" in reason:
return "kosdaq_scanner"
if "day_trade" in reason or "데이트레이딩" in reason:
return "day_trade"
if "rule" in reason or "룰" in reason:
return "rule_based"
return "unknown"
def _parse_exit_type(exit_reason: Optional[str]) -> str:
"""exit_reason 문자열에서 청산 유형 추출."""
if not exit_reason:
return "unknown"
reason = exit_reason.lower()
if "stop_loss" in reason or "손절" in reason:
return "stop_loss"
if "trailing" in reason or "트레일링" in reason:
return "trailing_stop"
if "take_profit" in reason or "익절" in reason:
return "take_profit"
if "eod" in reason or "장마감" in reason or "강제" in reason:
return "eod_forced"
if "manual" in reason or "수동" in reason:
return "manual"
if "rsi" in reason:
return "rsi_exit"
return "unknown"
# ─────────────────────────────────────────────────────────────────────────────
# 핵심 분석 함수
# ─────────────────────────────────────────────────────────────────────────────
async def analyze_performance(db: AsyncSession, days: int = 30) -> Dict[str, Any]:
"""
최근 N일 완료 거래 성과 분석.
Returns:
{
"summary": {전체 통계},
"by_strategy": {전략별 성과},
"by_weekday": {요일별 성과},
"by_exit_type": {청산 유형별 통계},
"worst_stocks": [최대 손실 종목 Top5],
"best_stocks": [최대 수익 종목 Top5],
"period_days": N,
"generated_at": ISO 문자열
}
"""
since = datetime.utcnow() - timedelta(days=days)
stmt = select(TradeHistory).where(
and_(
TradeHistory.order_type == "SELL",
TradeHistory.status == "FILLED",
TradeHistory.created_at >= since,
)
).order_by(TradeHistory.created_at)
result = await db.execute(stmt)
trades: List[TradeHistory] = result.scalars().all()
if not trades:
return {
"summary": _empty_summary(),
"by_strategy": {},
"by_weekday": {},
"by_exit_type": {},
"worst_stocks": [],
"best_stocks": [],
"period_days": days,
"total_trades": 0,
"generated_at": datetime.utcnow().isoformat(),
}
# ── 전체 통계 ──────────────────────────────────────────────────────────
pl_rates = [_to_float(t.profit_loss_rate) for t in trades]
pl_amounts = [_to_float(t.net_profit_loss) for t in trades]
wins = [r for r in pl_rates if r > 0]
losses = [r for r in pl_rates if r <= 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_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
# 최대 연속 손실
max_consecutive_loss = _calc_max_consecutive_loss(pl_rates)
summary = {
"total_trades": len(trades),
"win_trades": len(wins),
"loss_trades": len(losses),
"win_rate": round(win_rate * 100, 2),
"avg_win_pct": round(avg_win, 4),
"avg_loss_pct": round(avg_loss, 4),
"profit_factor": round(min(profit_factor, 99.0), 4),
"gross_profit": round(gross_profit, 2),
"gross_loss": round(gross_loss, 2),
"net_pnl": round(net_pnl, 2),
"max_consecutive_loss": max_consecutive_loss,
"risk_reward_ratio": round(abs(avg_win / avg_loss), 4) if avg_loss != 0 else 0.0,
}
# ── 전략별 성과 ──────────────────────────────────────────────────────
by_strategy: Dict[str, list] = {}
for t in trades:
strat = _parse_strategy(t.entry_reason)
by_strategy.setdefault(strat, []).append(_to_float(t.profit_loss_rate))
strategy_stats = {}
for strat, rates in by_strategy.items():
w = [r for r in rates if r > 0]
strategy_stats[strat] = {
"total": len(rates),
"win_rate": round(len(w) / len(rates) * 100, 2) if rates else 0.0,
"avg_return": round(sum(rates) / len(rates), 4) if rates else 0.0,
}
# ── 요일별 성과 ──────────────────────────────────────────────────────
weekday_names = ["월", "화", "수", "목", "금", "토", "일"]
by_weekday: Dict[str, list] = {}
for t in trades:
wday = weekday_names[t.created_at.weekday()] if t.created_at else "알수없음"
by_weekday.setdefault(wday, []).append(_to_float(t.profit_loss_rate))
weekday_stats = {
d: {
"total": len(rates),
"win_rate": round(len([r for r in rates if r > 0]) / len(rates) * 100, 2),
"avg_return": round(sum(rates) / len(rates), 4),
}
for d, rates in by_weekday.items()
if rates
}
# ── 청산 유형별 ──────────────────────────────────────────────────────
by_exit: Dict[str, list] = {}
for t in trades:
etype = _parse_exit_type(t.exit_reason)
by_exit.setdefault(etype, []).append(_to_float(t.profit_loss_rate))
exit_stats = {
etype: {
"count": len(rates),
"avg_return": round(sum(rates) / len(rates), 4),
"win_rate": round(len([r for r in rates if r > 0]) / len(rates) * 100, 2),
}
for etype, rates in by_exit.items()
if rates
}
# ── 종목별 집계 ──────────────────────────────────────────────────────
stock_pnl: Dict[str, float] = {}
for t in trades:
code = t.stock_code
stock_pnl[code] = stock_pnl.get(code, 0.0) + _to_float(t.net_profit_loss)
sorted_stocks = sorted(stock_pnl.items(), key=lambda x: x[1])
worst_stocks = [{"stock_code": c, "net_pnl": round(p, 2)} for c, p in sorted_stocks[:5]]
best_stocks = [{"stock_code": c, "net_pnl": round(p, 2)} for c, p in sorted_stocks[-5:][::-1]]
return {
"summary": summary,
"by_strategy": strategy_stats,
"by_weekday": weekday_stats,
"by_exit_type": exit_stats,
"worst_stocks": worst_stocks,
"best_stocks": best_stocks,
"period_days": days,
"total_trades": len(trades),
"generated_at": datetime.utcnow().isoformat(),
}
def _empty_summary() -> Dict[str, Any]:
return {
"total_trades": 0,
"win_trades": 0,
"loss_trades": 0,
"win_rate": 0.0,
"avg_win_pct": 0.0,
"avg_loss_pct": 0.0,
"profit_factor": 0.0,
"gross_profit": 0.0,
"gross_loss": 0.0,
"net_pnl": 0.0,
"max_consecutive_loss": 0,
"risk_reward_ratio": 0.0,
}
def _calc_max_consecutive_loss(pl_rates: List[float]) -> int:
max_streak = current_streak = 0
for r in pl_rates:
if r <= 0:
current_streak += 1
max_streak = max(max_streak, current_streak)
else:
current_streak = 0
return max_streak
# ─────────────────────────────────────────────────────────────────────────────
# 전략 인사이트
# ─────────────────────────────────────────────────────────────────────────────
def build_insights(perf: Dict[str, Any]) -> List[Dict[str, str]]:
"""
성과 데이터를 기반으로 개선 인사이트 목록을 생성.
Returns: [{"level": "warning"|"info"|"danger", "message": "..."}]
"""
insights = []
summary = perf.get("summary", {})
win_rate = summary.get("win_rate", 0.0)
pf = summary.get("profit_factor", 0.0)
avg_loss = summary.get("avg_loss_pct", 0.0)
max_cons_loss = summary.get("max_consecutive_loss", 0)
rr = summary.get("risk_reward_ratio", 0.0)
if win_rate < 40:
insights.append({"level": "danger", "message": f"승률 {win_rate:.1f}% — RSI 과매도 기준을 강화하거나 진입 조건을 추가 필터링하세요."})
elif win_rate < 50:
insights.append({"level": "warning", "message": f"승률 {win_rate:.1f}% — 50% 미만입니다. RSI_OVERSOLD_LEVEL 하향 조정을 검토하세요."})
else:
insights.append({"level": "info", "message": f"승률 {win_rate:.1f}% — 양호합니다."})
if 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} — 개선 여지 있습니다."})
if rr < 1.0 and rr > 0:
insights.append({"level": "warning", "message": f"손익비 {rr:.2f} — 1:1 미만입니다. ATR 익절 배수 또는 트레일링 스탑 완화를 검토하세요."})
if max_cons_loss >= 3:
insights.append({"level": "warning", "message": f"최대 연속 손실 {max_cons_loss}회 — 연속 손절 발생 중. 진입 필터(추세 필터, 시장국면 필터) 점검 필요."})
if avg_loss < -8.0:
insights.append({"level": "danger", "message": f"평균 손실 {avg_loss:.2f}% — 손절선({avg_loss:.2f}%)이 너무 넓습니다. STOP_LOSS_PERCENT 또는 ATR_STOP_MULTIPLIER 조정 권장."})
by_strategy = perf.get("by_strategy", {})
for strat, stats in by_strategy.items():
if stats.get("total", 0) >= 5 and stats.get("win_rate", 100) < 35:
insights.append({"level": "warning", "message": f"'{strat}' 전략 승률 {stats['win_rate']:.1f}% — 해당 전략 파라미터 재검토 권장."})
by_exit = perf.get("by_exit_type", {})
stop_loss_info = by_exit.get("stop_loss", {})
if stop_loss_info.get("count", 0) > 0:
sl_ratio = stop_loss_info["count"] / max(perf["total_trades"], 1)
if sl_ratio > 0.5:
insights.append({"level": "warning", "message": f"손절 청산 비율 {sl_ratio*100:.1f}% — 전체 청산의 절반 이상이 손절입니다. 진입 타이밍 재검토."})
return insights
# ─────────────────────────────────────────────────────────────────────────────
# 매매일지 생성
# ─────────────────────────────────────────────────────────────────────────────
async def get_uncovered_trades(db: AsyncSession, days: int = 7) -> List[TradeHistory]:
"""일지가 아직 없는 완료 SELL 거래 목록을 반환."""
since = datetime.utcnow() - timedelta(days=days)
# 이미 일지가 있는 trade_id 목록
covered_stmt = select(TradeJournal.trade_id).where(TradeJournal.trade_id.is_not(None))
covered_result = await db.execute(covered_stmt)
covered_ids = {row[0] for row in covered_result.fetchall()}
stmt = select(TradeHistory).where(
and_(
TradeHistory.order_type == "SELL",
TradeHistory.status == "FILLED",
TradeHistory.created_at >= since,
TradeHistory.id.not_in(covered_ids) if covered_ids else True,
)
).order_by(TradeHistory.created_at.desc())
result = await db.execute(stmt)
return result.scalars().all()
def generate_journal_for_trade(trade: TradeHistory, perf_context: Optional[Dict] = None) -> str:
"""
템플릿 기반 매매일지 내용 생성.
Args:
trade: 완료된 SELL TradeHistory 레코드
perf_context: 전체 성과 컨텍스트 (선택)
Returns:
일지 본문 문자열
"""
pl_rate = _to_float(trade.profit_loss_rate)
pl_amount = _to_float(trade.net_profit_loss)
exec_price = _to_float(trade.executed_price)
exec_qty = trade.executed_quantity or trade.order_quantity or 0
strategy = _parse_strategy(trade.entry_reason)
exit_type = _parse_exit_type(trade.exit_reason)
# 결과 평가
if pl_rate > 3.0:
result_label = "✅ 큰 수익 달성"
result_comment = "목표 수익을 충분히 달성했습니다. 진입 타이밍과 전략이 잘 맞았습니다."
elif pl_rate > 0:
result_label = "✅ 소폭 수익"
result_comment = "소폭 수익 실현. 목표 대비 아쉬운 결과이나 플러스 마감입니다."
elif pl_rate > -3.0:
result_label = "⚠️ 소폭 손실"
result_comment = "소폭 손실. 리스크 관리 규칙 내에서 처리되었습니다."
else:
result_label = "❌ 손절 처리"
result_comment = "손절 처리되었습니다. 진입 조건 및 시장 국면을 재검토하세요."
# 청산 유형별 코멘트
exit_comments = {
"stop_loss": "→ 손절 기준(-6% 또는 ATR 기반)에 도달하여 강제 청산.",
"trailing_stop": "→ 트레일링 스탑(고점-1.5×ATR)에 도달. 고점 대비 일정 비율 하락.",
"take_profit": "→ 익절 기준(+2R 또는 RSI 65+)에 도달하여 수익 확정.",
"eod_forced": "→ 장 마감 강제 청산(15:20 EOD). 포지션 오버나이트 방지.",
"rsi_exit": "→ RSI 과매수 구간 진입으로 부분/전량 청산.",
"manual": "→ 수동 청산 처리.",
"unknown": "→ 청산 사유 미상.",
}
exit_comment = exit_comments.get(exit_type, "→ 청산 완료.")
lines = [
f"# 매매일지 — {trade.stock_code} ({trade.stock_name or ''})",
f"**날짜**: {(trade.executed_at or trade.created_at).strftime('%Y-%m-%d %H:%M') if (trade.executed_at or trade.created_at) else '-'}",
f"**전략**: {strategy}",
"",
f"## 거래 결과: {result_label}",
f"- 체결가: {exec_price:,.0f}원 | 수량: {exec_qty:,}주",
f"- 손익률: **{pl_rate:+.2f}%** | 손익금액: {pl_amount:+,.0f}원",
f"- 청산 유형: `{exit_type}`",
f" {exit_comment}",
"",
"## 진입 근거",
f"{trade.entry_reason or '기록 없음'}",
"",
"## 청산 근거",
f"{trade.exit_reason or '기록 없음'}",
"",
"## 평가 및 교훈",
result_comment,
]
# 성과 컨텍스트 추가 (있을 경우)
if perf_context:
overall_wr = perf_context.get("summary", {}).get("win_rate", 0)
lines += [
"",
f"---",
f"*최근 {perf_context.get('period_days', 30)}일 전체 승률: {overall_wr:.1f}% (참고)*",
]
return "\n".join(lines)
async def create_journal_record(
db: AsyncSession,
trade: TradeHistory,
content: str,
) -> TradeJournal:
"""TradeJournal 레코드를 DB에 저장 후 반환."""
pl_rate = _to_float(trade.profit_loss_rate)
strategy = _parse_strategy(trade.entry_reason)
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")
journal = TradeJournal(
trade_id=trade.id,
stock_code=trade.stock_code,
stock_name=trade.stock_name,
journal_date=trade.executed_at or trade.created_at or datetime.utcnow(),
content=content,
strategy_name=strategy,
mood=mood,
tags=tags,
)
db.add(journal)
await db.flush()
return journal

View File

@@ -4,6 +4,7 @@ import TradeHistory from './pages/TradeHistory'
import MCPStatus from './pages/MCPStatus' import MCPStatus from './pages/MCPStatus'
import StrategySettings from './pages/StrategySettings' import StrategySettings from './pages/StrategySettings'
import Backtest from './pages/Backtest' import Backtest from './pages/Backtest'
import Analysis from './pages/Analysis'
import Layout from './components/Layout' import Layout from './components/Layout'
function App() { function App() {
@@ -15,6 +16,7 @@ function App() {
<Route path="/mcp" element={<MCPStatus />} /> <Route path="/mcp" element={<MCPStatus />} />
<Route path="/strategy" element={<StrategySettings />} /> <Route path="/strategy" element={<StrategySettings />} />
<Route path="/backtest" element={<Backtest />} /> <Route path="/backtest" element={<Backtest />} />
<Route path="/analysis" element={<Analysis />} />
</Routes> </Routes>
</Layout> </Layout>
) )

View File

@@ -1,5 +1,5 @@
import { Link, useLocation } from 'react-router-dom' import { Link, useLocation } from 'react-router-dom'
import { LayoutDashboard, History, Cpu, Activity, SlidersHorizontal, BarChart3 } from 'lucide-react' import { LayoutDashboard, History, Cpu, Activity, SlidersHorizontal, BarChart3, TrendingUp } from 'lucide-react'
import { useQuery } from '@tanstack/react-query' import { useQuery } from '@tanstack/react-query'
import api from '../lib/api' import api from '../lib/api'
import { useWebSocket } from '../hooks/useWebSocket' import { useWebSocket } from '../hooks/useWebSocket'
@@ -20,6 +20,7 @@ const Layout = ({ children }: LayoutProps) => {
const navItems = [ const navItems = [
{ path: '/', icon: LayoutDashboard, label: '대시보드' }, { path: '/', icon: LayoutDashboard, label: '대시보드' },
{ path: '/analysis', icon: TrendingUp, label: '성과 분석' },
{ path: '/strategy', icon: SlidersHorizontal, label: '전략 설정' }, { path: '/strategy', icon: SlidersHorizontal, label: '전략 설정' },
{ path: '/backtest', icon: BarChart3, label: '백테스트' }, { path: '/backtest', icon: BarChart3, label: '백테스트' },
{ path: '/trades', icon: History, label: '매매 로그' }, { path: '/trades', icon: History, label: '매매 로그' },

View File

@@ -0,0 +1,489 @@
import { useState } from 'react'
import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query'
import {
TrendingUp, TrendingDown, Target, AlertTriangle, CheckCircle, Info,
Play, RefreshCw, BookOpen, Zap, ChevronDown, ChevronUp
} from 'lucide-react'
import api from '../lib/api'
// ── 타입 정의 ─────────────────────────────────────────────────────────────────
interface PerformanceSummary {
total_trades: number
win_trades: number
loss_trades: number
win_rate: number
avg_win_pct: number
avg_loss_pct: number
profit_factor: number
gross_profit: number
gross_loss: number
net_pnl: number
max_consecutive_loss: number
risk_reward_ratio: number
}
interface PerformanceData {
summary: PerformanceSummary
by_strategy: Record<string, { total: number; win_rate: number; avg_return: number }>
by_exit_type: Record<string, { count: number; avg_return: number; win_rate: number }>
worst_stocks: { stock_code: string; net_pnl: number }[]
best_stocks: { stock_code: string; net_pnl: number }[]
period_days: number
total_trades: number
generated_at: string
}
interface Insight {
level: 'info' | 'warning' | 'danger'
message: string
}
interface OptimizationLog {
id: number
run_date: string
trigger: string
best_sharpe: number | null
current_sharpe: number | null
sharpe_improvement_pct: number | null
best_win_rate: number | null
best_mdd: number | null
applied: boolean
apply_reason: string
best_params: Record<string, unknown> | null
previous_params: Record<string, unknown> | null
total_combinations: number | null
created_at: string
}
// ── 유틸 함수 ─────────────────────────────────────────────────────────────────
const fmt = (n: number | null | undefined, digits = 2) =>
n == null ? '-' : n.toFixed(digits)
const fmtKRW = (n: number | null | undefined) =>
n == null ? '-' : `${n >= 0 ? '+' : ''}${n.toLocaleString('ko-KR')}원`
// ── 서브 컴포넌트: 통계 카드 ──────────────────────────────────────────────────
const StatCard = ({
label, value, sub, color = 'blue', icon: Icon,
}: {
label: string
value: string | number
sub?: string
color?: 'blue' | 'green' | 'red' | 'yellow' | 'purple'
icon?: React.ElementType
}) => {
const colorMap = {
blue: 'border-blue-500/30 bg-blue-500/5',
green: 'border-green-500/30 bg-green-500/5',
red: 'border-red-500/30 bg-red-500/5',
yellow: 'border-yellow-500/30 bg-yellow-500/5',
purple: 'border-purple-500/30 bg-purple-500/5',
}
const textMap = {
blue: 'text-blue-400', green: 'text-green-400', red: 'text-red-400',
yellow: 'text-yellow-400', purple: 'text-purple-400',
}
return (
<div className={`rounded-lg border p-4 ${colorMap[color]}`}>
<div className="flex items-center gap-2 mb-1">
{Icon && <Icon size={14} className={textMap[color]} />}
<span className="text-xs text-gray-400">{label}</span>
</div>
<div className={`text-2xl font-bold ${textMap[color]}`}>{value}</div>
{sub && <div className="text-xs text-gray-500 mt-0.5">{sub}</div>}
</div>
)
}
// ── 서브 컴포넌트: 인사이트 배지 ─────────────────────────────────────────────
const InsightBadge = ({ insight }: { insight: Insight }) => {
const map = {
info: { bg: 'bg-blue-500/10 border-blue-500/20', text: 'text-blue-300', Icon: Info },
warning: { bg: 'bg-yellow-500/10 border-yellow-500/20', text: 'text-yellow-300', Icon: AlertTriangle },
danger: { bg: 'bg-red-500/10 border-red-500/20', text: 'text-red-300', Icon: AlertTriangle },
}
const { bg, text, Icon } = map[insight.level]
return (
<div className={`flex items-start gap-2 p-3 rounded-lg border ${bg} text-sm`}>
<Icon size={14} className={`${text} mt-0.5 flex-shrink-0`} />
<span className={text}>{insight.message}</span>
</div>
)
}
// ── 메인 컴포넌트 ─────────────────────────────────────────────────────────────
export default function Analysis() {
const queryClient = useQueryClient()
const [days, setDays] = useState(30)
const [expandedOptLog, setExpandedOptLog] = useState<number | null>(null)
// 성과 데이터
const { data: perf, isLoading: perfLoading } = useQuery<PerformanceData>({
queryKey: ['analysis-performance', days],
queryFn: async () => (await api.get(`/analysis/performance?days=${days}`)).data,
refetchInterval: 60_000,
})
// 인사이트
const { data: insightsData, isLoading: insightsLoading } = useQuery<{
insights: Insight[]
last_optimization_date: string | null
}>({
queryKey: ['analysis-insights', days],
queryFn: async () => (await api.get(`/analysis/insights?days=${days}`)).data,
refetchInterval: 60_000,
})
// 최적화 이력
const { data: optHistory = [], isLoading: optHistLoading } = useQuery<OptimizationLog[]>({
queryKey: ['optimization-history'],
queryFn: async () => (await api.get('/analysis/optimization-history?limit=10')).data,
refetchInterval: 30_000,
})
// 최적화 상태
const { data: optStatus } = useQuery<{ running: boolean }>({
queryKey: ['optimization-status'],
queryFn: async () => (await api.get('/analysis/optimization-status')).data,
refetchInterval: 5_000,
})
// 일지 생성 뮤테이션
const generateJournals = useMutation({
mutationFn: () => api.post('/analysis/generate-journals?days=7'),
onSuccess: () => {
alert('일지 생성을 시작했습니다. 잠시 후 매매 로그에서 확인하세요.')
},
})
// 전략 최적화 뮤테이션
const optimizeStrategy = useMutation({
mutationFn: () =>
api.post(`/analysis/optimize-strategy?days=${days}&period_days=90`),
onSuccess: () => {
alert('전략 최적화를 시작했습니다. 수 분 후 최적화 이력에서 결과를 확인하세요.')
setTimeout(() => queryClient.invalidateQueries({ queryKey: ['optimization-history'] }), 10_000)
},
})
const summary = perf?.summary
const insights = insightsData?.insights || []
const isOptimizing = optStatus?.running || optimizeStrategy.isPending
return (
<div className="p-6 space-y-6 max-w-6xl mx-auto">
{/* 헤더 */}
<div className="flex items-center justify-between">
<div>
<h1 className="text-2xl font-bold text-white">성과 분석</h1>
<p className="text-sm text-gray-400 mt-1">매매 일지 자동 생성 & 전략 자동 최적화</p>
</div>
<div className="flex items-center gap-3">
<select
value={days}
onChange={(e) => setDays(Number(e.target.value))}
className="bg-gray-800 border border-gray-700 text-gray-300 rounded-md px-3 py-1.5 text-sm"
>
<option value={7}>최근 7일</option>
<option value={30}>최근 30일</option>
<option value={90}>최근 90일</option>
</select>
<button
onClick={() => queryClient.invalidateQueries({ queryKey: ['analysis-performance', days] })}
className="p-2 rounded-lg bg-gray-800 border border-gray-700 text-gray-400 hover:text-white transition-colors"
>
<RefreshCw size={16} />
</button>
</div>
</div>
{/* 성과 요약 카드 */}
{perfLoading ? (
<div className="text-gray-500 text-sm">성과 데이터 로딩 중...</div>
) : summary && summary.total_trades > 0 ? (
<div className="grid grid-cols-2 md:grid-cols-4 gap-4">
<StatCard
label="승률"
value={`${fmt(summary.win_rate)}%`}
sub={`${summary.win_trades}승 / ${summary.loss_trades}패`}
color={summary.win_rate >= 50 ? 'green' : summary.win_rate >= 40 ? 'yellow' : 'red'}
icon={Target}
/>
<StatCard
label="Profit Factor"
value={fmt(summary.profit_factor)}
sub="총수익 / 총손실"
color={summary.profit_factor >= 1.5 ? 'green' : summary.profit_factor >= 1.0 ? 'yellow' : 'red'}
icon={TrendingUp}
/>
<StatCard
label="손익비"
value={fmt(summary.risk_reward_ratio)}
sub={`평균수익 ${fmt(summary.avg_win_pct)}% / 평균손실 ${fmt(summary.avg_loss_pct)}%`}
color={summary.risk_reward_ratio >= 1.5 ? 'green' : summary.risk_reward_ratio >= 1.0 ? 'yellow' : 'red'}
icon={TrendingDown}
/>
<StatCard
label="순손익"
value={fmtKRW(summary.net_pnl)}
sub={`최대연속손실 ${summary.max_consecutive_loss}회`}
color={summary.net_pnl >= 0 ? 'green' : 'red'}
icon={summary.net_pnl >= 0 ? TrendingUp : TrendingDown}
/>
</div>
) : (
<div className="rounded-lg border border-gray-700 bg-gray-800/30 p-8 text-center text-gray-400">
최근 {days}일 내 완료된 매도 거래가 없습니다.
</div>
)}
<div className="grid grid-cols-1 lg:grid-cols-2 gap-6">
{/* 전략별 성과 */}
<div className="rounded-xl border border-gray-700 bg-gray-800/30 p-5">
<h2 className="text-base font-semibold text-white mb-4">전략별 성과</h2>
{perf && Object.keys(perf.by_strategy).length > 0 ? (
<div className="space-y-2">
{Object.entries(perf.by_strategy).map(([strat, stats]) => (
<div key={strat} className="flex items-center justify-between py-2 border-b border-gray-700/50 last:border-0">
<div>
<span className="text-sm text-gray-200 font-medium">{strat}</span>
<span className="ml-2 text-xs text-gray-500">{stats.total}건</span>
</div>
<div className="flex items-center gap-4 text-sm">
<span className={stats.win_rate >= 50 ? 'text-green-400' : 'text-red-400'}>
승률 {fmt(stats.win_rate)}%
</span>
<span className={stats.avg_return >= 0 ? 'text-green-400' : 'text-red-400'}>
평균 {stats.avg_return >= 0 ? '+' : ''}{fmt(stats.avg_return)}%
</span>
</div>
</div>
))}
</div>
) : (
<p className="text-gray-500 text-sm">데이터 없음</p>
)}
</div>
{/* 청산 유형별 */}
<div className="rounded-xl border border-gray-700 bg-gray-800/30 p-5">
<h2 className="text-base font-semibold text-white mb-4">청산 유형별 통계</h2>
{perf && Object.keys(perf.by_exit_type).length > 0 ? (
<div className="space-y-2">
{Object.entries(perf.by_exit_type).map(([etype, stats]) => (
<div key={etype} className="flex items-center justify-between py-2 border-b border-gray-700/50 last:border-0">
<div>
<span className="text-sm text-gray-200 font-medium">{etype}</span>
<span className="ml-2 text-xs text-gray-500">{stats.count}건</span>
</div>
<span className={`text-sm ${stats.avg_return >= 0 ? 'text-green-400' : 'text-red-400'}`}>
평균 {stats.avg_return >= 0 ? '+' : ''}{fmt(stats.avg_return)}%
</span>
</div>
))}
</div>
) : (
<p className="text-gray-500 text-sm">데이터 없음</p>
)}
</div>
</div>
{/* 인사이트 */}
<div className="rounded-xl border border-gray-700 bg-gray-800/30 p-5">
<h2 className="text-base font-semibold text-white mb-4 flex items-center gap-2">
<AlertTriangle size={16} className="text-yellow-400" />
전략 인사이트
</h2>
{insightsLoading ? (
<p className="text-gray-500 text-sm">분석 중...</p>
) : insights.length > 0 ? (
<div className="space-y-2">
{insights.map((ins, i) => <InsightBadge key={i} insight={ins} />)}
</div>
) : (
<p className="text-gray-500 text-sm">분석할 데이터가 없습니다.</p>
)}
</div>
{/* 매매일지 */}
<div className="rounded-xl border border-gray-700 bg-gray-800/30 p-5">
<div className="flex items-center justify-between mb-4">
<h2 className="text-base font-semibold text-white flex items-center gap-2">
<BookOpen size={16} className="text-blue-400" />
매매일지 자동 생성
</h2>
<button
onClick={() => generateJournals.mutate()}
disabled={generateJournals.isPending}
className="flex items-center gap-2 px-4 py-2 bg-blue-600 hover:bg-blue-700 disabled:bg-gray-700 text-white text-sm rounded-lg transition-colors"
>
{generateJournals.isPending ? (
<RefreshCw size={14} className="animate-spin" />
) : (
<Play size={14} />
)}
일지 자동 생성
</button>
</div>
<p className="text-sm text-gray-400">
최근 7일 내 일지가 없는 완료 거래에 대해 템플릿 기반 매매일지를 자동 생성합니다.
장 종료 후 16:00에 자동으로도 실행됩니다.
</p>
</div>
{/* 전략 자동 최적화 */}
<div className="rounded-xl border border-gray-700 bg-gray-800/30 p-5">
<div className="flex items-center justify-between mb-4">
<div>
<h2 className="text-base font-semibold text-white flex items-center gap-2">
<Zap size={16} className="text-purple-400" />
전략 자동 최적화
</h2>
{insightsData?.last_optimization_date && (
<p className="text-xs text-gray-500 mt-0.5">
마지막 최적화: {new Date(insightsData.last_optimization_date).toLocaleDateString('ko-KR')}
</p>
)}
</div>
<button
onClick={() => optimizeStrategy.mutate()}
disabled={isOptimizing}
className="flex items-center gap-2 px-4 py-2 bg-purple-600 hover:bg-purple-700 disabled:bg-gray-700 text-white text-sm rounded-lg transition-colors"
>
{isOptimizing ? (
<>
<RefreshCw size={14} className="animate-spin" />
최적화 중...
</>
) : (
<>
<Zap size={14} />
지금 최적화 실행
</>
)}
</button>
</div>
<p className="text-sm text-gray-400 mb-4">
RSI, 손절선, ATR 배수, 볼린저 밴드 파라미터를 그리드 서치로 최적화합니다.
Sharpe Ratio 10% 이상 개선 시 자동으로 전략 설정에 적용됩니다.
(매일 16:00 KST 자동 실행, 최근 7일 거래 5건 이상 시)
</p>
{/* 최적화 이력 */}
<div>
<h3 className="text-sm font-medium text-gray-300 mb-3">최적화 이력</h3>
{optHistLoading ? (
<p className="text-gray-500 text-sm">로딩 중...</p>
) : optHistory.length > 0 ? (
<div className="space-y-2">
{optHistory.map((log) => (
<div key={log.id} className="border border-gray-700 rounded-lg overflow-hidden">
<button
onClick={() => setExpandedOptLog(expandedOptLog === log.id ? null : log.id)}
className="w-full flex items-center justify-between p-3 hover:bg-gray-700/30 transition-colors"
>
<div className="flex items-center gap-3">
{log.applied ? (
<CheckCircle size={14} className="text-green-400" />
) : (
<Info size={14} className="text-gray-500" />
)}
<span className="text-sm text-gray-200">
{log.run_date}
</span>
<span className="text-xs text-gray-500">
{log.trigger === 'auto' ? '자동' : '수동'}
</span>
{log.applied && (
<span className="text-xs bg-green-500/20 text-green-400 px-2 py-0.5 rounded-full">
자동 적용됨
</span>
)}
</div>
<div className="flex items-center gap-4 text-sm">
{log.sharpe_improvement_pct != null && (
<span className={log.sharpe_improvement_pct >= 0 ? 'text-green-400' : 'text-red-400'}>
Sharpe {log.sharpe_improvement_pct >= 0 ? '+' : ''}{fmt(log.sharpe_improvement_pct)}%
</span>
)}
{expandedOptLog === log.id
? <ChevronUp size={14} className="text-gray-500" />
: <ChevronDown size={14} className="text-gray-500" />
}
</div>
</button>
{expandedOptLog === log.id && (
<div className="p-3 border-t border-gray-700 bg-gray-900/30 text-xs space-y-2">
<div className="grid grid-cols-3 gap-3">
<div>
<span className="text-gray-500">최적 Sharpe</span>
<p className="text-white">{fmt(log.best_sharpe, 3)}</p>
</div>
<div>
<span className="text-gray-500">최적 승률</span>
<p className="text-white">{fmt(log.best_win_rate, 1)}%</p>
</div>
<div>
<span className="text-gray-500">최적 MDD</span>
<p className="text-white">{fmt(log.best_mdd, 1)}%</p>
</div>
</div>
<div>
<span className="text-gray-500">사유: </span>
<span className="text-gray-300">{log.apply_reason || '-'}</span>
</div>
{log.best_params && (
<div>
<span className="text-gray-500 block mb-1">최적 파라미터:</span>
<pre className="text-gray-300 bg-gray-800 p-2 rounded text-xs overflow-x-auto">
{JSON.stringify(log.best_params, null, 2)}
</pre>
</div>
)}
</div>
)}
</div>
))}
</div>
) : (
<p className="text-gray-500 text-sm">최적화 이력이 없습니다.</p>
)}
</div>
</div>
{/* 최악/최고 종목 */}
{perf && (perf.worst_stocks.length > 0 || perf.best_stocks.length > 0) && (
<div className="grid grid-cols-1 md:grid-cols-2 gap-6">
<div className="rounded-xl border border-gray-700 bg-gray-800/30 p-5">
<h2 className="text-base font-semibold text-red-400 mb-3">최대 손실 종목 Top5</h2>
<div className="space-y-2">
{perf.worst_stocks.map((s) => (
<div key={s.stock_code} className="flex justify-between text-sm">
<span className="text-gray-200">{s.stock_code}</span>
<span className="text-red-400">{fmtKRW(s.net_pnl)}</span>
</div>
))}
</div>
</div>
<div className="rounded-xl border border-gray-700 bg-gray-800/30 p-5">
<h2 className="text-base font-semibold text-green-400 mb-3">최대 수익 종목 Top5</h2>
<div className="space-y-2">
{perf.best_stocks.map((s) => (
<div key={s.stock_code} className="flex justify-between text-sm">
<span className="text-gray-200">{s.stock_code}</span>
<span className="text-green-400">{fmtKRW(s.net_pnl)}</span>
</div>
))}
</div>
</div>
</div>
)}
</div>
)
}