shuim
a0973812fb
수개월간 매수 0건 + 백테스트가 손실로 보이던 근본 원인들을 수정. - ADX 발산(mean_reversion_strategy, run_backtest): Wilder 평활에 /period 누락으로 ADX가 100 초과 발산 → 실서비스 SIDEWAYS 게이트(ADX<20) 영구 무력화. 공식 복원(표준 Wilder와 대수적으로 동일 검증). - 백테스트 워밍업(run_backtest): 국면판정 220봉을 start_date부터 세서 14개월 미만 백테스트의 ~90%가 강제 BEAR → 가짜 손실. start_date 340일 전부터 프리로드하고 성과는 sim_start_date부터만 집계. CAGR 분모도 워밍업 제외한 실제 거래일로 수정. → 최근1년 -0.33%→+6.95%, 가산성 회복. - ML HOLD 게이트(trade_executor): 신뢰도 무관 매수 차단 → 데이터 부족으로 ML이 실패(conf=0)한 HOLD까지 모든 매수를 막던 것을 conf>=0.6 조건부로. - 빈 Redis 추천 캐시(swing/scheduler): 빈 캐시를 '히트'로 처리해 DB 폴백을 건너뛰던 것을 recommendations_by_rule 비면 DB 폴백하도록. - alembic 로깅(env.py, main.py): lifespan의 alembic upgrade가 fileConfig로 앱 로거를 꺼 startup 이후 로그 전멸 → 스킵(스키마는 create_all 담당). 검증: code-verifier 통과(실전 경로 CONFIRMED 버그 없음), walk-forward 6/9구간 양(+)·과적합 아님(조건부 견고). 서바이버십 편향은 별도 이슈로 검증 진행 중. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HBmzf1HDnvFHLraJ8eaaKJ
266 lines
12 KiB
Python
266 lines
12 KiB
Python
"""스윙 트레이딩 독립 스케줄러
|
|
|
|
main.py의 lifespan에서 asyncio.create_task()로 시작되는
|
|
백그라운드 태스크. 기존 auto_trade_scheduler()를 대체.
|
|
|
|
60초 주기 실행 흐름:
|
|
1. 장 오픈 체크 (장외 시 30분 대기)
|
|
2. 스윙 포지션 손절/익절 체크 (check_and_execute_exit_v2)
|
|
3. MDD 체크 + 강제 축소 실행 (10% → 30% 축소, 20% → 50% 축소)
|
|
4. AUTO_TRADE_ENABLED 체크
|
|
5. Redis 캐시 또는 DB에서 추천 조회
|
|
6. MCP RISK_ASSESSMENT → 승인 시 스윙 매수 실행
|
|
|
|
MDD 축소 (CLAUDE.md 5.3):
|
|
- reduction_applied == "10pct": 30% 축소 (REDUCE_AND_PAUSE)
|
|
- reduction_applied == "20pct": 50% 축소 (FORCE_REDUCE, 레버리지 ETF 우선)
|
|
|
|
추천 데이터 소스 (우선순위):
|
|
1. Redis: "recommendations:latest" 키 (배치 분석 결과 캐시)
|
|
2. DB: get_rule_based_recommendations_from_db() (Redis 미스 시 폴백)
|
|
"""
|
|
import asyncio
|
|
import logging
|
|
|
|
import pytz
|
|
from datetime import timedelta as _td
|
|
|
|
from app.core.config import settings
|
|
from app.core.database import AsyncSessionLocal
|
|
from app.core.market_schedule import is_market_open
|
|
from app.services.mdd_manager import mdd_manager
|
|
from app.services.swing.exit_service import check_and_execute_exit_v2
|
|
from app.services.swing.entry_service import execute_swing_entries, scan_dual_strategy_signals
|
|
|
|
logger = logging.getLogger(__name__)
|
|
_KST = pytz.timezone("Asia/Seoul")
|
|
|
|
|
|
def _normalize_recommendations(rec: dict) -> dict:
|
|
"""recommendations_by_rule 키 정규화.
|
|
|
|
Redis 캐시에 "recommendations" 키로 저장된 경우
|
|
"recommendations_by_rule"로 변환하여 execute_swing_entries()와 호환.
|
|
|
|
Args:
|
|
rec: 추천 딕셔너리
|
|
|
|
Returns:
|
|
dict: 정규화된 추천 딕셔너리
|
|
"""
|
|
if not isinstance(rec, dict):
|
|
return rec
|
|
if "recommendations_by_rule" in rec:
|
|
return rec
|
|
if "recommendations" in rec:
|
|
rec["recommendations_by_rule"] = rec["recommendations"]
|
|
return rec
|
|
|
|
|
|
def _filter_unsuitable_stocks(rec: dict, unsuitable_codes: set) -> None:
|
|
"""장 전 AI 분석에서 부적합 판정된 종목을 추천 목록에서 제거."""
|
|
groups = rec.get("recommendations_by_rule", [])
|
|
if isinstance(groups, list):
|
|
for group in groups:
|
|
if isinstance(group, dict) and "stocks" in group:
|
|
group["stocks"] = [
|
|
s for s in group["stocks"]
|
|
if isinstance(s, dict) and s.get("stock_code") not in unsuitable_codes
|
|
]
|
|
elif isinstance(groups, dict):
|
|
for key, stocks in groups.items():
|
|
if isinstance(stocks, list):
|
|
groups[key] = [
|
|
s for s in stocks
|
|
if isinstance(s, dict) and s.get("stock_code") not in unsuitable_codes
|
|
]
|
|
|
|
|
|
async def swing_scheduler() -> None:
|
|
"""60초 주기 스윙 트레이딩 스케줄러.
|
|
|
|
무한 루프로 실행되며 asyncio.CancelledError로만 종료.
|
|
각 사이클에서 예외 발생 시 로그 기록 후 다음 사이클 계속.
|
|
|
|
DB 세션은 사이클마다 AsyncSessionLocal()로 새로 생성하여
|
|
장시간 세션 유지로 인한 커넥션 풀 고갈 방지.
|
|
"""
|
|
await asyncio.sleep(2)
|
|
from app.api.endpoints.recommendations import (
|
|
get_rule_based_recommendations_from_db,
|
|
)
|
|
from app.api.endpoints.mcp import request_mcp_risk_assessment
|
|
from app.core.redis import get_json
|
|
|
|
logger.info("📈 [Swing] Scheduler started (every 60s)")
|
|
try:
|
|
while True:
|
|
try:
|
|
logger.info("[Swing] ── 사이클 시작 ──")
|
|
|
|
if not is_market_open():
|
|
logger.info("[Swing] 장 미오픈 — 30분 대기")
|
|
await asyncio.sleep(1800)
|
|
continue
|
|
|
|
# ── Phase 1: 청산 체크 (별도 세션) ──
|
|
async with AsyncSessionLocal() as exit_session:
|
|
try:
|
|
exits = await check_and_execute_exit_v2(exit_session)
|
|
if exits:
|
|
logger.info(f"[Swing] 청산 실행: {len(exits)}건")
|
|
except Exception as e:
|
|
logger.error(f"[Swing] 청산 체크 오류: {e}", exc_info=True)
|
|
|
|
# ── Phase 2: MDD 업데이트 + 체크 + 강제 축소 ──
|
|
# [FIX] mdd_manager.update()를 매 사이클 호출하여 MDD 감지 활성화
|
|
try:
|
|
from app.services.korea_investment_api import korea_investment_api
|
|
balance_resp = await korea_investment_api.get_balance()
|
|
if balance_resp:
|
|
_total = float(balance_resp.get("output2", [{}])[0].get("tot_evlu_amt", 0))
|
|
if _total > 0:
|
|
mdd_result = await mdd_manager.update(_total)
|
|
if mdd_result.get("action") != "NONE":
|
|
logger.warning(f"[Swing] MDD 상태 변경: {mdd_result}")
|
|
except Exception as e:
|
|
logger.debug(f"[Swing] MDD update 조회 실패 (무시): {e}")
|
|
|
|
can_open_new = await mdd_manager.can_open_new_position()
|
|
logger.info(f"[Swing] MDD 체크: can_open={can_open_new}")
|
|
if not can_open_new:
|
|
logger.warning("[Swing] MDD 초과 — 신규 진입 차단")
|
|
if mdd_manager.reduction_applied:
|
|
try:
|
|
reduce_pct = 0.50 if mdd_manager.reduction_applied == "20pct" else 0.30
|
|
mdd_action = {
|
|
"action": "FORCE_REDUCE" if mdd_manager.reduction_applied == "20pct" else "REDUCE_AND_PAUSE",
|
|
"reduce_pct": reduce_pct,
|
|
"prioritize_leverage": mdd_manager.reduction_applied == "20pct",
|
|
}
|
|
results = await mdd_manager.execute_reduction(mdd_action)
|
|
if results:
|
|
logger.warning(f"[Swing] MDD 강제 축소 완료: {len(results)}종목")
|
|
except Exception as e:
|
|
logger.error(f"[Swing] MDD 강제 축소 실패: {e}", exc_info=True)
|
|
await asyncio.sleep(60)
|
|
continue
|
|
|
|
# 3. 자동매매 활성화 여부
|
|
if not settings.AUTO_TRADE_ENABLED:
|
|
logger.info("[Swing] AUTO_TRADE_ENABLED=False — 스킵")
|
|
await asyncio.sleep(60)
|
|
continue
|
|
|
|
# ── Phase 3: 추천 조회 + 진입 실행 (별도 세션) ──
|
|
# 4. Redis 캐시 확인 → cache miss 시 DB 조회
|
|
rec = await get_json("recommendations:latest")
|
|
# [FIX] Redis 캐시가 존재해도 recommendations_by_rule이 비어있으면(배치가 빈
|
|
# 추천을 캐싱한 경우) 기존엔 '히트'로 처리해 DB 폴백을 건너뛰고 스킵했다.
|
|
# 빈 캐시가 30분 TTL 동안 DB의 유효 후보를 가려 매매를 막으므로, 비어있으면
|
|
# 캐시 미스와 동일하게 DB 조회로 폴백한다.
|
|
if rec and rec.get("recommendations_by_rule"):
|
|
rec = _normalize_recommendations(rec)
|
|
logger.info(f"[Swing] Redis 캐시 히트 (추천 {len(rec.get('recommendations_by_rule', []))}그룹)")
|
|
else:
|
|
logger.info("[Swing] Redis 캐시 미스/빈값 → DB 조회")
|
|
async with AsyncSessionLocal() as rec_session:
|
|
rec = await get_rule_based_recommendations_from_db(rec_session)
|
|
|
|
if not rec or not rec.get("recommendations_by_rule"):
|
|
logger.info("[Swing] 추천 데이터 없음 — 스킵")
|
|
await asyncio.sleep(60)
|
|
continue
|
|
|
|
# 5. 장 전 AI 분석 결과 활용 + MCP 리스크 평가 (최적화)
|
|
# 장 전 Sonnet 배치 분석 결과가 Redis에 있으면 부적합 종목을 사전 필터링
|
|
# MCP RISK_ASSESSMENT는 실제 매수 직전에만 호출 (비용 절감)
|
|
_ai_filtered_codes = set()
|
|
try:
|
|
_ai_pre = await get_json("ai:pre_market:latest")
|
|
if _ai_pre and isinstance(_ai_pre, dict):
|
|
for s in _ai_pre.get("suitable_stocks", []):
|
|
if isinstance(s, dict) and not s.get("suitable", True):
|
|
_ai_filtered_codes.add(s.get("stock_code", ""))
|
|
if _ai_filtered_codes:
|
|
logger.info(f"[Swing] AI 장 전 분석 필터: {len(_ai_filtered_codes)}종목 제외")
|
|
except Exception:
|
|
pass
|
|
|
|
# 추천 목록에서 AI 부적합 종목 제거
|
|
if _ai_filtered_codes:
|
|
_filter_unsuitable_stocks(rec, _ai_filtered_codes)
|
|
|
|
recommendations = rec.get("recommendations_by_rule", [])
|
|
if isinstance(recommendations, dict):
|
|
rec_count = sum(len(v) for v in recommendations.values())
|
|
top_stocks = [
|
|
item.get("stock_code", "")
|
|
for items in recommendations.values()
|
|
for item in (items[:3] if isinstance(items, list) else [])
|
|
][:10]
|
|
else:
|
|
rec_count = sum(
|
|
len(g.get("stocks", [])) for g in recommendations
|
|
if isinstance(g, dict)
|
|
)
|
|
top_stocks = [
|
|
s.get("stock_code", "")
|
|
for g in recommendations if isinstance(g, dict)
|
|
for s in g.get("stocks", [])[:3]
|
|
][:10]
|
|
|
|
if not top_stocks:
|
|
logger.info("[Swing] 필터링 후 추천 종목 없음 — 스킵")
|
|
await asyncio.sleep(60)
|
|
continue
|
|
|
|
# MCP 리스크 평가 (Haiku — 실제 진입 직전 1회)
|
|
trade_plan = {
|
|
"action": "BUY",
|
|
"strategy": "swing",
|
|
"total_recommendations": rec_count,
|
|
"top_stocks": top_stocks,
|
|
}
|
|
|
|
logger.info(f"[Swing] MCP 리스크 평가 요청: {len(top_stocks)}종목")
|
|
mcp_approved, mcp_reason = await request_mcp_risk_assessment(
|
|
trade_plan, {"strategy": "swing"}, {"top_stocks": top_stocks}
|
|
)
|
|
logger.info(f"[Swing] MCP 결과: approved={mcp_approved}, reason={mcp_reason}")
|
|
if not mcp_approved:
|
|
logger.warning(f"[Swing] MCP 거부: {mcp_reason}")
|
|
from app.services.db_worker import enqueue_db_write
|
|
await enqueue_db_write({
|
|
"type": "log_mcp_deny",
|
|
"reason": mcp_reason,
|
|
"trade_plan": trade_plan,
|
|
})
|
|
await asyncio.sleep(60)
|
|
continue
|
|
|
|
# 6. 스윙 진입 실행 (별도 세션)
|
|
async with AsyncSessionLocal() as entry_session:
|
|
logger.info(f"[Swing] 진입 실행 시작: {top_stocks[:5]}")
|
|
await execute_swing_entries(rec, entry_session)
|
|
|
|
# 7. 듀얼 전략 시그널 스캔 (모멘텀 돌파 + 눌림목 매수, CLAUDE.md 8.4)
|
|
try:
|
|
async with AsyncSessionLocal() as dual_session:
|
|
dual_results = await scan_dual_strategy_signals(dual_session)
|
|
if dual_results:
|
|
logger.info(f"[Swing] 듀얼 전략 진입: {len(dual_results)}건")
|
|
except Exception as e:
|
|
logger.debug(f"[Swing] 듀얼 전략 스캔 오류: {e}")
|
|
|
|
logger.info("[Swing] ── 사이클 완료 ──")
|
|
|
|
except Exception as e:
|
|
logger.error(f"[Swing] 스케줄러 사이클 오류: {e}", exc_info=True)
|
|
|
|
await asyncio.sleep(60)
|
|
|
|
except asyncio.CancelledError:
|
|
logger.info("[Swing] Scheduler stopped")
|
|
raise
|