shuim
9fd02932fa
모든 스케줄러가 장 마감 중 고정 1800초를 자서 09:00 개장을 최대 30분 건너뛰었고, 배치 분석은 개장 후에야 시작돼 추천 종목이 09:00에 비어 있었음. - market_schedule.is_pre_market_window() 추가 (08:30~09:00, 거래일만) - batch_analysis_scheduler: 개장 전 준비 구간에도 분석 실행 → stock_analysis·Redis 추천 캐시를 미리 채워 09:00에 추천 즉시 표시 - batch/auto_trade/swing 스케줄러: 마감 중 sleep = min(1800, 개장까지 남은 시간) - 테스트 1건 추가 (총 5건) Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
750 lines
35 KiB
Python
750 lines
35 KiB
Python
"""
|
|
배치 분석, 장 종료 후 분석, KRX 데이터 수집 스케줄러 모듈
|
|
|
|
- batch_analysis_scheduler(): 10분 간격 배치 분석 (stock_analysis DB 업데이트)
|
|
- post_market_analyzer_scheduler(): 장 종료 후 매매일지 생성 + 전략 최적화
|
|
- krx_data_collector_scheduler(): 장 종료 후 일봉 데이터 수집
|
|
- _normalize_recommendations(): 추천 결과 포맷 정규화 헬퍼
|
|
"""
|
|
import asyncio
|
|
import logging
|
|
import pytz
|
|
from datetime import datetime as _dt, date as _date, timedelta as _td
|
|
|
|
from app.core.config import settings
|
|
from app.core.database import AsyncSessionLocal
|
|
from app.core.market_schedule import (
|
|
is_market_open, seconds_until_market_open, is_pre_market_window,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _normalize_recommendations(rec: dict) -> dict:
|
|
"""서로 다른 포맷의 추천 결과를 recommendations_by_rule 포맷으로 정규화.
|
|
|
|
- get_rule_based_recommendations_realtime() → {"recommendations": [...], ...}
|
|
- get_rule_based_recommendations_from_db() → {"recommendations_by_rule": {...}, ...}
|
|
|
|
둘 다 recommendations_by_rule 키를 보장한 상태로 돌려준다.
|
|
Redis 캐시 데이터와 DB 데이터 모두 이 함수를 통해 정규화.
|
|
변환 실패 시 원본 rec를 그대로 반환 (런타임 에러 방지).
|
|
"""
|
|
if not rec:
|
|
return rec or {}
|
|
|
|
try:
|
|
# Case 1: Already normalized (dict)
|
|
if isinstance(rec.get("recommendations_by_rule"), dict):
|
|
return rec
|
|
|
|
# Determine the source of the recommendation list
|
|
raw_list = []
|
|
if "recommendations" in rec and isinstance(rec["recommendations"], list):
|
|
raw_list = rec["recommendations"]
|
|
elif "recommendations_by_rule" in rec and isinstance(rec["recommendations_by_rule"], list):
|
|
# This is the case that causes the bug
|
|
raw_list = rec["recommendations_by_rule"]
|
|
|
|
if not raw_list:
|
|
# if there is no list to process, make sure the output has a dict
|
|
rec["recommendations_by_rule"] = {}
|
|
return rec
|
|
|
|
# Group the list by rule
|
|
by_rule: dict = {}
|
|
for item in raw_list:
|
|
rule = (
|
|
item.get("rule_name")
|
|
or item.get("signal")
|
|
or item.get("strategy")
|
|
or "general"
|
|
)
|
|
by_rule.setdefault(rule, []).append(item)
|
|
|
|
# Return the original dict with the new/replaced 'recommendations_by_rule'
|
|
return {**rec, "recommendations_by_rule": by_rule}
|
|
|
|
except Exception as e:
|
|
logger.warning(f"⚠️ _normalize_recommendations 변환 실패, 원본 반환: {e}")
|
|
return rec
|
|
|
|
|
|
async def batch_analysis_scheduler():
|
|
"""10분 간격 배치 분석 스케줄러.
|
|
|
|
장중 10분마다 KIS API로 종목 분석 후 stock_analysis DB에 upsert.
|
|
분석 완료 후 Redis에 recommendations:latest 캐시를 저장해
|
|
auto_trade_scheduler가 DB 조회 없이 바로 매매 판단에 활용할 수 있도록 한다.
|
|
|
|
개선:
|
|
- 15:30 이후 장 종료 감지 → 익일 08:50까지 자동 중단
|
|
- 분석 실패 시 exponential backoff 재시도 (최대 3회: 10s/20s/40s)
|
|
"""
|
|
await asyncio.sleep(15) # 부팅 안정화 (auto_trade_scheduler보다 늦게 시작)
|
|
from app.api.endpoints.recommendations import get_rule_based_recommendations_realtime
|
|
from app.core.redis import setex_json, TTL_RECOMMENDATIONS, TTL_ANALYSIS
|
|
|
|
interval = int(getattr(settings, "BATCH_ANALYSIS_INTERVAL", 600)) # 기본 10분
|
|
_KST = pytz.timezone("Asia/Seoul")
|
|
logger.info(f"📊 Batch analysis scheduler started (every {interval}s)")
|
|
try:
|
|
while True:
|
|
now_kst = pytz.datetime.datetime.now(_KST)
|
|
|
|
# 15:30 이후 → 장 종료, 익일 08:50까지 대기
|
|
if now_kst.hour > 15 or (now_kst.hour == 15 and now_kst.minute >= 30):
|
|
tomorrow_start = (
|
|
_KST.localize(
|
|
pytz.datetime.datetime(now_kst.year, now_kst.month, now_kst.day, 8, 50, 0)
|
|
) + _td(days=1)
|
|
)
|
|
sleep_sec = max((tomorrow_start - now_kst).total_seconds(), 3600)
|
|
logger.info(f"⏸️ [Batch] 장 종료 (15:30 이후). {sleep_sec / 3600:.1f}시간 후 재개")
|
|
await asyncio.sleep(sleep_sec)
|
|
continue
|
|
|
|
# [FIX 2026-09-23] 개장 전 준비 구간(08:30~09:00)에는 분석을 미리 돌려
|
|
# stock_analysis·Redis 추천 캐시를 채워둔다 → 09:00에 추천 종목 즉시 표시.
|
|
# 그 외 마감 시간에는 개장을 넘겨 자지 않도록 sleep 상한을 둔다.
|
|
if not is_market_open() and not is_pre_market_window():
|
|
_until_open = seconds_until_market_open()
|
|
await asyncio.sleep(1800.0 if _until_open is None else max(15.0, min(1800.0, _until_open)))
|
|
continue
|
|
|
|
# ── exponential backoff 재시도 (최대 3회) ────────────────────────
|
|
max_retries = 3
|
|
for attempt in range(max_retries):
|
|
try:
|
|
suffix = f" (재시도 {attempt + 1}/{max_retries})" if attempt > 0 else ""
|
|
logger.info(f"🔄 [Batch] 10-min analysis cycle starting...{suffix}")
|
|
result = await get_rule_based_recommendations_realtime()
|
|
count = len(result.get("recommendations", [])) if result else 0
|
|
logger.info(f"✅ [Batch] Done: {count} stocks saved/updated in stock_analysis DB")
|
|
|
|
# ── Redis 캐시 업데이트 ─────────────────────────────────────────
|
|
if result:
|
|
# 정규화: recommendations_by_rule 키 보장 (auto_trade_scheduler 호환)
|
|
normalized = _normalize_recommendations(result)
|
|
ok = await setex_json(
|
|
"recommendations:latest", TTL_RECOMMENDATIONS, normalized
|
|
)
|
|
if ok:
|
|
rec_by_rule = normalized.get('recommendations_by_rule', [])
|
|
rec_count = len(rec_by_rule) if isinstance(rec_by_rule, (list, dict)) else 0
|
|
logger.info(
|
|
f"💾 [Batch] Redis 캐시 갱신 완료: recommendations:latest "
|
|
f"(rule 수: {rec_count}, "
|
|
f"TTL: {TTL_RECOMMENDATIONS}s)"
|
|
)
|
|
|
|
# ===== 개별 종목별 캐시 (병렬 처리 적용) =====
|
|
raw_list = result.get("recommendations", [])
|
|
cache_tasks = []
|
|
|
|
for item in raw_list:
|
|
code = item.get("stock_code") or item.get("code")
|
|
if code:
|
|
cache_tasks.append(setex_json(f"analysis:{code}", TTL_ANALYSIS, item))
|
|
|
|
if cache_tasks:
|
|
await asyncio.gather(*cache_tasks, return_exceptions=True)
|
|
logger.info(f"💾 [Batch] 개별 종목 분석 데이터 캐시 갱신 완료: {len(cache_tasks)}건 (TTL: {TTL_ANALYSIS}s)")
|
|
# ===============================================
|
|
# ───────────────────────────────────────────────────────
|
|
break # 성공 시 재시도 루프 탈출
|
|
|
|
except Exception as e:
|
|
wait = 10 * (2 ** attempt) # 10s → 20s → 40s
|
|
if attempt < max_retries - 1:
|
|
logger.warning(
|
|
f"⚠️ [Batch] 분석 실패 ({attempt + 1}/{max_retries}), "
|
|
f"{wait}초 후 재시도: {e}"
|
|
)
|
|
await asyncio.sleep(wait)
|
|
else:
|
|
logger.error(f"❌ Batch analysis error (최대 재시도 초과): {e}", exc_info=True)
|
|
|
|
await asyncio.sleep(interval)
|
|
except asyncio.CancelledError:
|
|
logger.info("⏹️ Batch analysis scheduler stopped")
|
|
raise
|
|
|
|
|
|
async def post_market_analyzer_scheduler():
|
|
"""장 종료 후 16:00 KST 1회 실행 — 매매일지 자동 생성 + 전략 자동 최적화.
|
|
|
|
조건:
|
|
- 매일 16:00 KST 에 1회 실행 (이미 오늘 실행했으면 스킵)
|
|
- 최근 7일 내 SELL FILLED 거래 >= 5건이면 최적화 실행
|
|
- 마지막 최적화로부터 3일 이상 경과 시만 최적화 실행
|
|
"""
|
|
_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 select, 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)
|
|
# [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:
|
|
logger.info(f"📝 [PostMarket] 일지 {journal_count}건 큐 등록 완료")
|
|
|
|
# ── Telegram 일일 성과 리포트 ──────────────────────────────────
|
|
try:
|
|
from app.services.telegram_notifier import telegram_notifier
|
|
# 당일 SELL 체결 내역으로 PnL 계산
|
|
_today_start = _utcnow_dt.utcnow().replace(hour=0, minute=0, second=0, microsecond=0)
|
|
_today_sells = await db.execute(
|
|
select(TradeHistory).where(
|
|
and_(
|
|
TradeHistory.order_type == "SELL",
|
|
TradeHistory.status.in_(["COMPLETED", "FILLED"]),
|
|
TradeHistory.created_at >= _today_start,
|
|
)
|
|
)
|
|
)
|
|
_sells = _today_sells.scalars().all()
|
|
_total_pnl = sum(float(t.net_profit_loss or 0) for t in _sells)
|
|
_win = sum(1 for t in _sells if float(t.net_profit_loss or 0) > 0)
|
|
_loss = sum(1 for t in _sells if float(t.net_profit_loss or 0) <= 0)
|
|
# 포트폴리오 요약
|
|
_pf_value, _cash = 0.0, 0.0
|
|
try:
|
|
from app.api.endpoints.portfolio import portfolio_cache
|
|
_summary = portfolio_cache.get_summary_copy()
|
|
if _summary:
|
|
_pf_value = float(_summary.get("total_assets", 0))
|
|
_cash = float(_summary.get("available_cash", 0))
|
|
except Exception:
|
|
pass
|
|
await telegram_notifier.notify_daily_report(
|
|
date_str=str(today),
|
|
total_pnl=_total_pnl,
|
|
trade_count=len(_sells),
|
|
win_count=_win,
|
|
loss_count=_loss,
|
|
portfolio_value=_pf_value,
|
|
cash_balance=_cash,
|
|
)
|
|
except Exception as _tg_err:
|
|
logger.debug(f"[PostMarket] Telegram 리포트 전송 실패: {_tg_err}")
|
|
|
|
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.in_(["COMPLETED", "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():
|
|
"""장 종료 후 KRX 일봉 데이터 자동 수집 스케줄러.
|
|
|
|
매일 15:40 KST 이후 collect_all(update_mode=True) 를 1회 실행.
|
|
이미 오늘 수집 완료된 경우 다음 날까지 스킵.
|
|
"""
|
|
_KST = pytz.timezone("Asia/Seoul")
|
|
_last_collected: _date | None = None
|
|
|
|
await asyncio.sleep(30) # 부팅 안정화
|
|
logger.info("📅 KRX data collector scheduler started (daily after 15:40 KST)")
|
|
|
|
try:
|
|
while True:
|
|
now_kst = _dt.now(_KST)
|
|
today = now_kst.date()
|
|
|
|
# 오늘 이미 수집했으면 자정까지 대기
|
|
if _last_collected == today:
|
|
# 다음 날 16:00까지 남은 초 계산
|
|
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()
|
|
logger.info(f"📅 KRX 오늘 수집 완료. 다음 수집까지 {sleep_sec/3600:.1f}시간 대기")
|
|
await asyncio.sleep(max(sleep_sec, 60))
|
|
continue
|
|
|
|
# 15:40 이전이면 대기
|
|
collect_hour, collect_min = 15, 40
|
|
if (now_kst.hour, now_kst.minute) < (collect_hour, collect_min):
|
|
# 오늘 15:40까지 남은 초
|
|
target = _KST.localize(
|
|
_dt(today.year, today.month, today.day, collect_hour, collect_min, 0)
|
|
)
|
|
sleep_sec = (target - now_kst).total_seconds()
|
|
if sleep_sec > 0:
|
|
logger.info(f"📅 KRX 수집 대기 중... 15:40까지 {sleep_sec/60:.0f}분 남음")
|
|
await asyncio.sleep(min(sleep_sec, 1800)) # 최대 30분마다 재확인
|
|
continue
|
|
|
|
# 주말(토/일)은 스킵하고 다음 날 재확인
|
|
if today.weekday() >= 5:
|
|
await asyncio.sleep(3600)
|
|
continue
|
|
|
|
# 수집 실행 (최대 3회 재시도, 30분 간격)
|
|
max_krx_retries = 3
|
|
for krx_attempt in range(max_krx_retries):
|
|
try:
|
|
suffix = f" (재시도 {krx_attempt + 1}/{max_krx_retries})" if krx_attempt > 0 else ""
|
|
logger.info(f"📥 [KRX] 일봉 데이터 증분 수집 시작 (update mode){suffix}...")
|
|
from collect_krx_data import collect_all
|
|
await collect_all(update_mode=True)
|
|
_last_collected = today
|
|
logger.info("✅ [KRX] 일봉 데이터 수집 완료")
|
|
break
|
|
except Exception as e:
|
|
if krx_attempt < max_krx_retries - 1:
|
|
logger.warning(
|
|
f"⚠️ [KRX] 수집 실패 ({krx_attempt + 1}/{max_krx_retries}), "
|
|
f"30분 후 재시도: {e}"
|
|
)
|
|
await asyncio.sleep(1800)
|
|
else:
|
|
logger.error(f"❌ [KRX] 데이터 수집 최대 재시도 초과: {e}", exc_info=True)
|
|
|
|
except asyncio.CancelledError:
|
|
logger.info("⏹️ KRX data collector scheduler stopped")
|
|
raise
|
|
|
|
|
|
# Deprecated schedulers (kept for backwards compatibility)
|
|
|
|
async def auto_trade_scheduler():
|
|
"""[DEPRECATED] 1분 간격 자동매매 스케줄러.
|
|
|
|
.. deprecated::
|
|
services/swing/scheduler.py의 swing_scheduler()로 대체됨.
|
|
main.py lifespan에서 더 이상 호출하지 않음.
|
|
하위 호환을 위해 함수는 유지.
|
|
"""
|
|
import warnings
|
|
warnings.warn(
|
|
"auto_trade_scheduler is deprecated. Use services/swing/scheduler.py instead.",
|
|
DeprecationWarning,
|
|
stacklevel=2,
|
|
)
|
|
# 원래 기능: Redis 캐시 확인 후 매매 실행 (swing_scheduler로 대체됨)
|
|
await asyncio.sleep(2)
|
|
from app.api.endpoints.recommendations import (
|
|
get_rule_based_recommendations_from_db,
|
|
execute_auto_trade_background,
|
|
)
|
|
from app.api.endpoints.portfolio import check_and_execute_exit_trades
|
|
from app.api.endpoints.mcp import request_mcp_risk_assessment
|
|
from app.core.redis import get_json
|
|
|
|
logger.info("⏱️ Auto-trade scheduler started (every 60s, Redis→DB fallback mode)")
|
|
try:
|
|
while True:
|
|
try:
|
|
if not is_market_open():
|
|
# 개장을 넘겨 자지 않도록 상한 (09:00부터 바로 매매 판단 시작)
|
|
_until_open = seconds_until_market_open()
|
|
_sleep = 1800.0 if _until_open is None else max(15.0, min(1800.0, _until_open))
|
|
logger.info(f"💤 Market is closed. Next check in {_sleep / 60:.0f} minutes.")
|
|
await asyncio.sleep(_sleep)
|
|
continue
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
# 1. 보유 종목 손절/익절 체크
|
|
logger.info("💰 Checking portfolio for stop-loss/take-profit...")
|
|
try:
|
|
await check_and_execute_exit_trades(session)
|
|
except Exception as e:
|
|
logger.error(f"Exit trade check error: {e}", exc_info=True)
|
|
|
|
# 2. MDD 체크 (CLAUDE.md 5.3) — 신규 진입 가능 여부
|
|
from app.services.mdd_manager import mdd_manager
|
|
can_open_new = await mdd_manager.can_open_new_position()
|
|
if not can_open_new:
|
|
logger.warning(
|
|
"⚠️ [Auto-Trade] MDD -10% 이상 도달 — 신규 진입 차단. "
|
|
"기존 포지션만 손절/익절 모니터링 중"
|
|
)
|
|
|
|
# 3. 자동매매: Redis 캐시 우선 → cache miss 시만 DB 조회
|
|
if settings.AUTO_TRADE_ENABLED and can_open_new:
|
|
# ── Redis 캐시 확인 (10분 배치 결과, TTL 30분) ────────────────
|
|
rec = await get_json("recommendations:latest")
|
|
if rec:
|
|
rec_by_rule = rec.get('recommendations_by_rule', [])
|
|
rec_count = len(rec_by_rule) if isinstance(rec_by_rule, (list, dict)) else 0
|
|
logger.info(
|
|
"⚡ Redis cache hit: recommendations:latest "
|
|
f"(rule 수: {rec_count}) "
|
|
"→ DB 조회 생략"
|
|
)
|
|
# 포맷 정규화 (다른 키 조합 대응)
|
|
rec = _normalize_recommendations(rec)
|
|
else:
|
|
logger.info("🔄 Redis cache miss → DB에서 추천 조회")
|
|
rec = await get_rule_based_recommendations_from_db(session)
|
|
# ──────────────────────────────────────────────────────
|
|
|
|
if rec and rec.get("recommendations_by_rule"):
|
|
# ── MCP RISK_ASSESSMENT (공통 함수, 이슈 4) ──────────
|
|
recommendations = rec.get("recommendations_by_rule", [])
|
|
|
|
# Handle both list and dict formats for backwards compatibility
|
|
if isinstance(recommendations, dict):
|
|
# Old format: {rule_name: [stocks]}
|
|
recommendation_count = sum(len(v) for v in recommendations.values())
|
|
rules_list = list(recommendations.keys())
|
|
top_stocks_list = [
|
|
item.get("stock_code") or item.get("code", "")
|
|
for rule_items in recommendations.values()
|
|
for item in (
|
|
rule_items[:3]
|
|
if isinstance(rule_items, list)
|
|
else []
|
|
)
|
|
][:10]
|
|
else:
|
|
# New format: [{rule_id, rule_name, stocks: [...]}]
|
|
recommendation_count = sum(
|
|
len(rule_group.get("stocks", []))
|
|
for rule_group in recommendations
|
|
if isinstance(rule_group, dict)
|
|
)
|
|
rules_list = [
|
|
rule_group.get("rule_name", rule_group.get("rule_id", ""))
|
|
for rule_group in recommendations
|
|
if isinstance(rule_group, dict)
|
|
]
|
|
top_stocks_list = [
|
|
item.get("stock_code") or item.get("code", "")
|
|
for rule_group in recommendations
|
|
if isinstance(rule_group, dict)
|
|
for item in rule_group.get("stocks", [])[:3]
|
|
][:10]
|
|
|
|
trade_plan = {
|
|
"action": "BUY",
|
|
"strategy": "rule_based_batch",
|
|
"recommendation_count": recommendation_count,
|
|
"rules": rules_list,
|
|
"top_stocks": top_stocks_list,
|
|
}
|
|
portfolio: dict = {}
|
|
signals = {
|
|
"source": "db_recommendations",
|
|
"rules_triggered": rules_list,
|
|
}
|
|
approve, reason = await request_mcp_risk_assessment(
|
|
trade_plan, portfolio, signals
|
|
)
|
|
# ─────────────────────────────────────────────────────
|
|
if not approve:
|
|
logger.warning(
|
|
f"🛑 MCP RISK_ASSESSMENT DENY → 주문 스킵. 사유: {reason}"
|
|
)
|
|
try:
|
|
from app.services.db_worker import enqueue_db_write
|
|
await enqueue_db_write({
|
|
"type": "log_mcp_deny",
|
|
"reason": reason,
|
|
"trade_plan": trade_plan,
|
|
})
|
|
except Exception:
|
|
pass
|
|
else:
|
|
if reason:
|
|
logger.info(
|
|
f"✅ MCP RISK_ASSESSMENT APPROVE. 참고: {reason}"
|
|
)
|
|
logger.info("📊 Found recommendations, executing auto-trade")
|
|
await execute_auto_trade_background(rec, session)
|
|
else:
|
|
logger.info("⏭️ Auto-trade skipped: no recommendations")
|
|
elif settings.AUTO_TRADE_ENABLED and not can_open_new:
|
|
logger.info("⏸️ Auto-trade disabled (MDD protection) - only monitoring stop-loss/take-profit")
|
|
else:
|
|
logger.info("⏸️ Auto-trade disabled - only monitoring stop-loss/take-profit")
|
|
except Exception as e:
|
|
logger.error(f"❌ Auto-trade scheduler error: {e}", exc_info=True)
|
|
|
|
await asyncio.sleep(60)
|
|
except asyncio.CancelledError:
|
|
logger.info("⏹️ Auto-trade scheduler stopped")
|
|
raise
|
|
|
|
|
|
async def day_trade_scheduler():
|
|
"""[DEPRECATED] 30초 간격 데이트레이딩 스케줄러.
|
|
|
|
.. deprecated::
|
|
services/daytrading/scheduler.py의 daytrading_scheduler()로 대체됨.
|
|
main.py lifespan에서 더 이상 호출하지 않음.
|
|
하위 호환을 위해 함수는 유지.
|
|
"""
|
|
# 원래 기능: ORB 수집, 신호 스캔, EOD 청산 (daytrading_scheduler로 대체됨)
|
|
await asyncio.sleep(5)
|
|
from app.api.endpoints.day_trading import run_day_trading_cycle
|
|
from app.services.mdd_manager import mdd_manager
|
|
|
|
logger.info("📈 Day-trade scheduler started (every 30s)")
|
|
try:
|
|
while True:
|
|
if not is_market_open():
|
|
await asyncio.sleep(1800)
|
|
continue
|
|
if not settings.DAY_TRADE_ENABLED:
|
|
await asyncio.sleep(60)
|
|
continue
|
|
|
|
# MDD 체크: 단기 전략 활성화 여부 확인 (CLAUDE.md 5.3)
|
|
short_term_enabled = await mdd_manager.is_short_term_strategy_enabled()
|
|
if not short_term_enabled:
|
|
logger.info("⏸️ [DayTrade] MDD -10% 이상 — 단기 전략(데이트레이딩) 중단 (신규 진입 불가)")
|
|
await asyncio.sleep(60) # 1분 후 재확인
|
|
continue
|
|
|
|
try:
|
|
result = await run_day_trading_cycle()
|
|
phase = result.get("phase", "")
|
|
if result.get("entries"):
|
|
logger.info(f"📈 데이트레이딩 진입: {result['entries']}")
|
|
if result.get("exits"):
|
|
logger.info(f"📉 데이트레이딩 청산: {result['exits']}")
|
|
if phase not in ("market_closed", "outside_window"):
|
|
logger.debug(f"Day-trade cycle [{phase}]: {result.get('time')}")
|
|
except Exception as e:
|
|
logger.error(f"❌ Day-trade scheduler error: {e}", exc_info=True)
|
|
await asyncio.sleep(30)
|
|
except asyncio.CancelledError:
|
|
logger.info("⏹️ Day-trade scheduler stopped")
|
|
raise
|
|
|
|
|
|
async def news_collect_scheduler():
|
|
"""뉴스 수집 스케줄러 (NEWS_COLLECT_INTERVAL 간격, 기본 30분).
|
|
|
|
- 장중에만 수집 (is_market_open)
|
|
- 시장 뉴스 + 활성 포트폴리오 종목 뉴스 수집
|
|
- NEWS_ENABLED=False 시 비활성화
|
|
"""
|
|
await asyncio.sleep(20) # 부팅 안정화 대기
|
|
|
|
if not settings.NEWS_ENABLED:
|
|
logger.info("News collect scheduler disabled (NEWS_ENABLED=False)")
|
|
return
|
|
|
|
logger.info(
|
|
f"News collect scheduler started "
|
|
f"(every {settings.NEWS_COLLECT_INTERVAL}s)"
|
|
)
|
|
|
|
try:
|
|
while True:
|
|
if not is_market_open():
|
|
await asyncio.sleep(300) # 장 외 5분 대기
|
|
continue
|
|
|
|
try:
|
|
from app.services.news_sentiment import (
|
|
get_market_news_score,
|
|
get_stock_news_score,
|
|
)
|
|
|
|
# 1. 시장 뉴스 수집
|
|
market_score = await get_market_news_score()
|
|
logger.info(
|
|
f"[NEWS_SCHEDULER] market_score={market_score:.3f}"
|
|
)
|
|
|
|
# 2. 활성 포트폴리오 종목 뉴스 수집
|
|
try:
|
|
from sqlalchemy import select
|
|
from app.core.database import AsyncSessionLocal
|
|
from app.models.models import Portfolio
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
result = await session.execute(
|
|
select(Portfolio.stock_code).where(
|
|
Portfolio.quantity > 0
|
|
)
|
|
)
|
|
held_codes = [row[0] for row in result.fetchall()]
|
|
|
|
for code in held_codes[:20]: # 최대 20종목
|
|
try:
|
|
score = await get_stock_news_score(code)
|
|
if score <= settings.NEWS_REDUCE_SCORE:
|
|
logger.warning(
|
|
f"[NEWS_SCHEDULER] {code} 악재 감지: "
|
|
f"score={score:.3f}"
|
|
)
|
|
except Exception as e:
|
|
logger.debug(
|
|
f"[NEWS_SCHEDULER] {code} 뉴스 수집 오류: {e}"
|
|
)
|
|
await asyncio.sleep(1) # 크롤링 간격 (서버 부하 방지)
|
|
|
|
except Exception as e:
|
|
logger.warning(
|
|
f"[NEWS_SCHEDULER] 포트폴리오 종목 뉴스 수집 오류: {e}"
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(
|
|
f"[NEWS_SCHEDULER] 뉴스 수집 사이클 오류: {e}",
|
|
exc_info=True,
|
|
)
|
|
|
|
await asyncio.sleep(settings.NEWS_COLLECT_INTERVAL)
|
|
|
|
except asyncio.CancelledError:
|
|
logger.info("News collect scheduler stopped")
|
|
raise
|
|
|
|
|
|
async def pre_market_us_scheduler():
|
|
"""07:00 KST 미장 심리 분석 스케줄러 (1회/일).
|
|
|
|
전날 밤 미국 시장(S&P500, NASDAQ, VIX) 종가를 기반으로
|
|
한국 장 개장 전 적응형 전략 모드를 보정한다.
|
|
|
|
[실행 시점] 매일 07:00 KST (미장 마감 06:00 이후, 한국 장 09:00 이전)
|
|
[소요 시간] 약 3~5초 (네이버 3페이지 + Yahoo 1건)
|
|
"""
|
|
_KST = pytz.timezone("Asia/Seoul")
|
|
_last_run: _date | None = None
|
|
|
|
await asyncio.sleep(10) # 부팅 안정화
|
|
logger.info("🌍 Pre-market US sentiment scheduler started (daily 07:00 KST)")
|
|
|
|
try:
|
|
while True:
|
|
now_kst = _dt.now(_KST)
|
|
today = now_kst.date()
|
|
|
|
# 오늘 이미 실행했으면 내일 07:00까지 대기
|
|
if _last_run == today:
|
|
tomorrow_07 = _KST.localize(
|
|
_dt(today.year, today.month, today.day, 7, 0, 0)
|
|
) + _td(days=1)
|
|
sleep_sec = max((tomorrow_07 - now_kst).total_seconds(), 3600)
|
|
await asyncio.sleep(sleep_sec)
|
|
continue
|
|
|
|
# 주말(토/일) 스킵
|
|
if today.weekday() >= 5:
|
|
await asyncio.sleep(3600)
|
|
continue
|
|
|
|
# 07:00 이전이면 대기
|
|
if now_kst.hour < 7:
|
|
target_07 = _KST.localize(
|
|
_dt(today.year, today.month, today.day, 7, 0, 0)
|
|
)
|
|
sleep_sec = (target_07 - now_kst).total_seconds()
|
|
if sleep_sec > 0:
|
|
await asyncio.sleep(min(sleep_sec, 1800))
|
|
continue
|
|
|
|
# 09:00 이후면 오늘은 스킵 (이미 장 시작)
|
|
if now_kst.hour >= 9:
|
|
_last_run = today
|
|
await asyncio.sleep(3600)
|
|
continue
|
|
|
|
# 실행
|
|
try:
|
|
from app.services.us_market_sentiment import run_pre_market_check
|
|
result = await run_pre_market_check()
|
|
_last_run = today
|
|
logger.info(
|
|
"✅ [PreMarket] 미장 심리 분석 완료: %s (override=%s)",
|
|
result.get("sentiment", "?"),
|
|
result.get("regime_override") or "없음",
|
|
)
|
|
except Exception as e:
|
|
logger.error("❌ [PreMarket] 미장 심리 분석 실패: %s", e, exc_info=True)
|
|
# 실패해도 하루 1회만 시도 (무한 재시도 방지)
|
|
_last_run = today
|
|
|
|
await asyncio.sleep(3600)
|
|
|
|
except asyncio.CancelledError:
|
|
logger.info("⏹️ Pre-market US sentiment scheduler stopped")
|
|
raise
|