diff --git a/score-engine/main.py b/score-engine/main.py index 0d4c613..55ee935 100644 --- a/score-engine/main.py +++ b/score-engine/main.py @@ -526,17 +526,27 @@ def calc_short_score(short_data: list) -> tuple[float, str]: # ── H1: 5년 재무 추세 점수 ──────────────────────────────── -async def calc_trend_score(conn, stock_code: str) -> tuple[float, str]: +async def calc_trend_score(conn, stock_code: str, as_of: date | None = None) -> tuple[float, str]: """ 최근 5년치 사업보고서 ROE/영업이익률의 일관성·추세 점수 (-30~+30) - 버핏: 안정적이고 우상향하는 수익성 선호 + 버핏: 안정적이고 우상향하는 수익성 선호. + as_of=date면 그 시점 이전 공시된 보고서만 사용 (look-ahead bias 차단). """ - rows = await conn.fetch(""" - SELECT bsns_year, roe, operating_margin - FROM dart_financials - WHERE stock_code=$1 AND reprt_code='11011' AND roe IS NOT NULL - ORDER BY bsns_year DESC LIMIT 5 - """, stock_code) + if as_of is None: + rows = await conn.fetch(""" + SELECT bsns_year, roe, operating_margin + FROM dart_financials + WHERE stock_code=$1 AND reprt_code='11011' AND roe IS NOT NULL + ORDER BY bsns_year DESC LIMIT 5 + """, stock_code) + else: + rows = await conn.fetch(f""" + SELECT bsns_year, roe, operating_margin + FROM dart_financials + WHERE stock_code=$1 AND reprt_code='11011' AND roe IS NOT NULL + AND ((bsns_year::int + 1)::text || '-04-01')::date <= $2 + ORDER BY bsns_year DESC LIMIT 5 + """, stock_code, as_of) if len(rows) < 2: return 0.0, "" roes = [float(r["roe"]) for r in rows] @@ -753,15 +763,22 @@ def calc_peg(curr: dict, prev: dict, per: float) -> tuple[float, str, str]: # ── 퀄리티+모멘텀 (12-1개월 가격 모멘텀) ────────────────── -async def calc_momentum(conn, stock_code: str) -> tuple[float, str, str]: +async def calc_momentum(conn, stock_code: str, as_of: date | None = None) -> tuple[float, str, str]: """ AQR 스타일 12-1개월 모멘텀: (P_t-21 / P_t-252) - 1 - 최근 1개월 제외(반전효과 회피)한 11개월 수익률 + 최근 1개월 제외(반전효과 회피)한 11개월 수익률. + as_of=date면 그 시점 이전 가격만 사용. """ - rows = await conn.fetch(""" - SELECT close_price, dt FROM stock_ohlcv - WHERE stock_code=$1 ORDER BY dt DESC LIMIT 260 - """, stock_code) + if as_of is None: + rows = await conn.fetch(""" + SELECT close_price, dt FROM stock_ohlcv + WHERE stock_code=$1 ORDER BY dt DESC LIMIT 260 + """, stock_code) + else: + rows = await conn.fetch(""" + SELECT close_price, dt FROM stock_ohlcv + WHERE stock_code=$1 AND dt <= $2 ORDER BY dt DESC LIMIT 260 + """, stock_code, as_of) if len(rows) < 200: return 0.0, "관망", "" closes = [(r["dt"], float(r["close_price"])) for r in rows if r["close_price"] > 0] @@ -885,16 +902,23 @@ async def calc_mohanram_g(conn, stock_code: str, sector: str, fin_curr: dict, fi # ── Amihud 비유동성 (2002) ──────────────────────────────── -async def calc_amihud(conn, stock_code: str) -> tuple[float, str, str]: +async def calc_amihud(conn, stock_code: str, as_of: date | None = None) -> tuple[float, str, str]: """ Amihud (2002): ILLIQ = avg(|return| / 거래대금) × 1e9 소형주 비유동성 프리미엄 — 높을수록 알파 잠재력 ↑ but 거래 어려움 - 20일 평균 사용 (1년 미만 데이터에서도 작동) + 20일 평균 사용 (1년 미만 데이터에서도 작동). + as_of=date면 그 시점 이전 가격/거래량만 사용. """ - rows = await conn.fetch(""" - SELECT close_price, volume FROM stock_ohlcv - WHERE stock_code=$1 ORDER BY dt DESC LIMIT 21 - """, stock_code) + if as_of is None: + rows = await conn.fetch(""" + SELECT close_price, volume FROM stock_ohlcv + WHERE stock_code=$1 ORDER BY dt DESC LIMIT 21 + """, stock_code) + else: + rows = await conn.fetch(""" + SELECT close_price, volume FROM stock_ohlcv + WHERE stock_code=$1 AND dt <= $2 ORDER BY dt DESC LIMIT 21 + """, stock_code, as_of) if len(rows) < 10: return 0.0, "관망", "" closes = [float(r["close_price"]) for r in rows if r["close_price"] > 0] vols = [int(r["volume"] or 0) for r in rows] @@ -916,19 +940,29 @@ async def calc_amihud(conn, stock_code: str) -> tuple[float, str, str]: # ── 시장 베타 (BAB 핵심 — Frazzini-Pedersen 2014) ────────── -async def calc_beta(conn, stock_code: str, days: int = 60) -> tuple[float, str, str]: +async def calc_beta(conn, stock_code: str, days: int = 60, as_of: date | None = None) -> tuple[float, str, str]: """ 종목 일별 수익률 vs KOSPI 60일 회귀 베타 BAB(Betting Against Beta) 알파: 저베타 종목이 위험조정 후 우월 - β < 0.7 매수 (저베타 알파), β > 1.5 매도 (고베타 위험), 그 사이 관망 + β < 0.7 매수 (저베타 알파), β > 1.5 매도 (고베타 위험), 그 사이 관망. + as_of=date면 그 시점 이전 가격만 사용. """ - rows = await conn.fetch(""" - SELECT s.dt, s.close_price AS stk, k.close_price AS kospi - FROM stock_ohlcv s - JOIN stock_ohlcv k ON k.dt=s.dt AND k.stock_code='KOSPI' - WHERE s.stock_code=$1 - ORDER BY s.dt DESC LIMIT $2 - """, stock_code, days + 1) + if as_of is None: + rows = await conn.fetch(""" + SELECT s.dt, s.close_price AS stk, k.close_price AS kospi + FROM stock_ohlcv s + JOIN stock_ohlcv k ON k.dt=s.dt AND k.stock_code='KOSPI' + WHERE s.stock_code=$1 + ORDER BY s.dt DESC LIMIT $2 + """, stock_code, days + 1) + else: + rows = await conn.fetch(""" + SELECT s.dt, s.close_price AS stk, k.close_price AS kospi + FROM stock_ohlcv s + JOIN stock_ohlcv k ON k.dt=s.dt AND k.stock_code='KOSPI' + WHERE s.stock_code=$1 AND s.dt <= $3 + ORDER BY s.dt DESC LIMIT $2 + """, stock_code, days + 1, as_of) if len(rows) < 30: return 0.0, "관망", "" s_rets, k_rets = [], [] for i in range(len(rows) - 1): @@ -1035,20 +1069,65 @@ def _similar_weight(similar_count: int | None) -> float: return min(2.5, _math.sqrt(n)) +# 사후 학습된 reliability/credibility 캐시 (calculate_daily_scores 시작 시 1회 로드) +_RELIABILITY_CACHE: dict = {} # {(catalyst, time_horizon): reliability_score} +_SRC_CRED_CACHE: dict = {} # {source: credibility} — DB 학습값 (sample≥20) + +async def _load_reliability_caches(conn, backfill_mode: bool = False): + """일 1회 호출 — 사후 검증 잡이 채운 신뢰도 테이블을 메모리 캐시로 로드. + backfill_mode=True면 캐시를 비워둠 (사후 학습된 신뢰도를 과거 시점에 적용하면 look-ahead bias). + """ + _RELIABILITY_CACHE.clear() + _SRC_CRED_CACHE.clear() + if backfill_mode: + return + try: + for r in await conn.fetch( + "SELECT catalyst, time_horizon, reliability_score, sample_size " + "FROM sentiment_reliability"): + if (r["sample_size"] or 0) >= 5: + _RELIABILITY_CACHE[(r["catalyst"], r["time_horizon"])] = float(r["reliability_score"] or 1.0) + for r in await conn.fetch( + "SELECT source, credibility, sample_size FROM news_source_credibility"): + if (r["sample_size"] or 0) >= 20: + _SRC_CRED_CACHE[r["source"]] = float(r["credibility"] or 0.5) + except Exception as e: + logger.warning("reliability.cache.err", error=str(e)) + + async def calc_news_score_weighted( conn, stock_code: str, week_ago: date, now: datetime | None = None ) -> tuple[float, dict]: """ - catalyst×intensity×시간감쇠×중복가중 적용된 뉴스 점수 (-100~+100). - 동시에 attention/neutral 카운트도 함께 반환 (호재·악재만 점수에 반영). + 종합 가중치 = catalyst × intensity × 시간감쇠 × 중복가중 + × 출처신뢰도 × 제목강도 × LLM신뢰도 × 사후신뢰도(reliability) + × 사건시드여부(첫 뉴스만 풀가중·후속 0.30) + × stock_impacts 매핑(1뉴스 N종목 영향도) + → -100~+100. 동시에 pos/neg/neutral 카운트 반환. + + primary_stock 외에 stock_impacts에 등장하는 종목도 가중치 비례 점수 포함. """ now = now or datetime.now(timezone.utc) week_ago_dt = datetime.combine(week_ago, datetime.min.time(), tzinfo=timezone.utc) + # primary_stock=stock OR stock_impacts에 stock_code가 키로 등장하는 뉴스 전체 + # 시간감쇠 기준: published_at (없으면 analyzed_at). 일부 RSS가 오래된 기사 재노출 → + # analyzed_at만 보면 11개월 전 기사도 풀가중 → 점수 왜곡. published_at 정렬·필터링 필수. rows = await conn.fetch(""" SELECT sentiment, intensity, COALESCE(catalyst, '기타') AS catalyst, - analyzed_at, COALESCE(similar_count, 1) AS sim + analyzed_at, + COALESCE(published_at, analyzed_at) AS ref_at, + COALESCE(similar_count, 1) AS sim, + COALESCE(time_horizon, '단기') AS time_horizon, + COALESCE(impact_scope, '종목') AS impact_scope, + COALESCE(llm_confidence, 0.5) AS llm_confidence, + COALESCE(source_credibility, 0.5) AS source_credibility, + COALESCE(title_strength, 1.0) AS title_strength, + COALESCE(is_event_seed, TRUE) AS is_event_seed, + COALESCE(stock_impacts, '{}'::jsonb) AS stock_impacts, + source FROM news_analysis - WHERE primary_stock=$1 AND analyzed_at >= $2 + WHERE (primary_stock=$1 OR stock_impacts ? $1) + AND COALESCE(published_at, analyzed_at) >= $2 """, stock_code, week_ago_dt) if not rows: return 0.0, {"pos": 0, "neg": 0, "neutral": 0, "total": 0} @@ -1244,16 +1323,22 @@ async def fetch_naver_ohlcv(conn, code: str, days: int = 400) -> int: # ── H3: 시장 레짐 ───────────────────────────────────────── -async def calc_market_regime(conn) -> tuple[str, float]: +async def calc_market_regime(conn, as_of: date | None = None) -> tuple[str, float]: """ KOSPI 종가 vs 200일 이평으로 시장 레짐 판단 위면 강세(+5), 아래면 약세(-10), 데이터 없으면 중립 + as_of=date이면 그 시점 기준 (백필용). """ - # KOSPI 인덱스를 stock_ohlcv에 코드 'KOSPI'로 저장한다고 가정 - rows = await conn.fetch(""" - SELECT close_price FROM stock_ohlcv - WHERE stock_code='KOSPI' ORDER BY dt DESC LIMIT 200 - """) + if as_of is None: + rows = await conn.fetch(""" + SELECT close_price FROM stock_ohlcv + WHERE stock_code='KOSPI' ORDER BY dt DESC LIMIT 200 + """) + else: + rows = await conn.fetch(""" + SELECT close_price FROM stock_ohlcv + WHERE stock_code='KOSPI' AND dt <= $1 ORDER BY dt DESC LIMIT 200 + """, as_of) if len(rows) < 100: return "데이터부족", 0.0 closes = [float(r["close_price"]) for r in rows if r["close_price"] > 0] @@ -1297,28 +1382,55 @@ def is_value_investable(fin: dict, per: float, pbr: float, market_cap: int) -> t return False, f"부채비율 {fin.get('debt_ratio',0):.0f}% 초과" # PER>60 단독 배제는 폐지(B) — 성장/테마주 구제 위해 호출부에서 # PEG·12-1모멘텀과 함께 판단 (calculate_daily_scores, 모멘텀 산출 직후) - if market_cap > 0 and market_cap < 10_000_000_000: # 100억 미만 - return False, "시총 100억 미만" + if market_cap > 0 and market_cap < 30_000_000_000: # 300억 미만 (잡주 제외) + return False, "시총 300억 미만 (잡주)" return True, "" # ── 일간 점수 산출 ──────────────────────────────────────── +def _disclosure_date_sql() -> str: + """DART 보고서 종류별 표준 공시일 추정 SQL 표현식. + 한국 공시 규정: 사업보고서(11011)는 사업연도 종료 후 90일 이내(다음 해 3/31), + 분기/반기는 분기 종료 후 45일 이내. 안전하게 1일 여유 둠. + 백필 모드에서 'estimated_disclosure_date <= as_of' 필터에 사용해 미래 보고서 누설 차단.""" + return ("""(CASE reprt_code + WHEN '11011' THEN ((bsns_year::int + 1)::text || '-04-01')::date + WHEN '11012' THEN (bsns_year || '-05-16')::date + WHEN '11013' THEN (bsns_year || '-08-15')::date + WHEN '11014' THEN (bsns_year || '-11-15')::date + ELSE '9999-12-31'::date + END)""") + + # ══════════════════════════════════════════════════════════ # 신규 보조 시그널 (임원매매 / 컨센서스 / 매크로 / 기관 / 밸류 percentile) # ══════════════════════════════════════════════════════════ -async def _load_insider_map(conn) -> dict: - """최근 90일 임원·대주주 매매 집계. 종목별 (net_change, buy_cnt, sell_cnt, top_actor)""" - rows = await conn.fetch(""" - SELECT stock_code, - SUM(shares_change) AS net, - SUM(CASE WHEN shares_change > 0 THEN 1 ELSE 0 END) AS buys, - SUM(CASE WHEN shares_change < 0 THEN 1 ELSE 0 END) AS sells, - COUNT(*) AS total - FROM dart_insider_trades - WHERE rcept_dt >= CURRENT_DATE - 90 - GROUP BY stock_code - """) +async def _load_insider_map(conn, as_of: date | None = None) -> dict: + """최근 90일 임원·대주주 매매 집계. 종목별 (net_change, buy_cnt, sell_cnt, top_actor). + as_of=None이면 CURRENT_DATE 기준, as_of=date이면 그 시점 기준 (백필용).""" + if as_of is None: + rows = await conn.fetch(""" + SELECT stock_code, + SUM(shares_change) AS net, + SUM(CASE WHEN shares_change > 0 THEN 1 ELSE 0 END) AS buys, + SUM(CASE WHEN shares_change < 0 THEN 1 ELSE 0 END) AS sells, + COUNT(*) AS total + FROM dart_insider_trades + WHERE rcept_dt >= CURRENT_DATE - 90 + GROUP BY stock_code + """) + else: + rows = await conn.fetch(""" + SELECT stock_code, + SUM(shares_change) AS net, + SUM(CASE WHEN shares_change > 0 THEN 1 ELSE 0 END) AS buys, + SUM(CASE WHEN shares_change < 0 THEN 1 ELSE 0 END) AS sells, + COUNT(*) AS total + FROM dart_insider_trades + WHERE rcept_dt BETWEEN $1::date - 90 AND $1::date + GROUP BY stock_code + """, as_of) return {r["stock_code"]: dict(r) for r in rows} @@ -1341,7 +1453,11 @@ def calc_insider_signal(stat: dict) -> tuple[float, str]: return sig, reason -async def _load_consensus_map(conn) -> dict: +async def _load_consensus_map(conn, as_of: date | None = None) -> dict: + """as_of=None이면 최근 30일 컨센서스. as_of=date(백필)이면 빈 dict 반환 + (analyst_consensus는 최신값만 저장하고 시점 이력이 없어서 look-ahead bias 위험).""" + if as_of is not None: + return {} rows = await conn.fetch( "SELECT stock_code, target_price, recomm_mean FROM analyst_consensus " "WHERE updated_at >= CURRENT_DATE - 30") @@ -1374,13 +1490,20 @@ def calc_consensus_signal(cons: dict, cur_price: float) -> tuple[float, str]: return sig, " ".join(reason_parts) -async def _load_macro_state(conn) -> dict: - """최근 5일 vs 그 이전 5일 매크로 변동률""" - rows = await conn.fetch(""" - SELECT indicator, trade_date, value FROM macro_daily - WHERE trade_date >= CURRENT_DATE - 20 - ORDER BY indicator, trade_date DESC - """) +async def _load_macro_state(conn, as_of: date | None = None) -> dict: + """최근 5일 vs 그 이전 5일 매크로 변동률. as_of=date면 그 시점 기준 20일 윈도우.""" + if as_of is None: + rows = await conn.fetch(""" + SELECT indicator, trade_date, value FROM macro_daily + WHERE trade_date >= CURRENT_DATE - 20 + ORDER BY indicator, trade_date DESC + """) + else: + rows = await conn.fetch(""" + SELECT indicator, trade_date, value FROM macro_daily + WHERE trade_date BETWEEN $1::date - 20 AND $1::date + ORDER BY indicator, trade_date DESC + """, as_of) by_ind: dict = {} for r in rows: by_ind.setdefault(r["indicator"], []).append(float(r["value"])) @@ -1436,17 +1559,28 @@ def calc_macro_signal(sector: str, macro: dict) -> tuple[float, str]: return sig, " ".join(parts) -async def _load_inst_flow_map(conn) -> dict: - """종목별 최근 5일 기관·외국인 순매수 합계""" - rows = await conn.fetch(""" - SELECT stock_code, - SUM(inst_net) AS inst5d, - SUM(foreign_net) AS for5d, - AVG(close_price)::float AS avg_price - FROM inst_daily_flow - WHERE trade_date >= CURRENT_DATE - 7 - GROUP BY stock_code - """) +async def _load_inst_flow_map(conn, as_of: date | None = None) -> dict: + """종목별 최근 5일 기관·외국인 순매수 합계. as_of=date면 그 시점 기준.""" + if as_of is None: + rows = await conn.fetch(""" + SELECT stock_code, + SUM(inst_net) AS inst5d, + SUM(foreign_net) AS for5d, + AVG(close_price)::float AS avg_price + FROM inst_daily_flow + WHERE trade_date >= CURRENT_DATE - 7 + GROUP BY stock_code + """) + else: + rows = await conn.fetch(""" + SELECT stock_code, + SUM(inst_net) AS inst5d, + SUM(foreign_net) AS for5d, + AVG(close_price)::float AS avg_price + FROM inst_daily_flow + WHERE trade_date BETWEEN $1::date - 7 AND $1::date + GROUP BY stock_code + """, as_of) return {r["stock_code"]: dict(r) for r in rows} @@ -1494,41 +1628,64 @@ def calc_valuation_percentile(per_history: list, cur_per: float) -> tuple[float, return 0.0, "" -async def calculate_daily_scores(): - logger.info("scoring.start") - today = date.today() +async def calculate_daily_scores(as_of: date | None = None): + """일간 점수 계산. as_of=None이면 today (운영 모드), as_of=date이면 그 시점 기준 (백필 모드). + 백필 모드는 look-ahead bias 차단: + - 사후 학습 캐시(reliability/source_credibility) 미적용 + - weight_config는 config_date <= as_of 필터 + - 컨센서스(이력 추적 없음) 제외 + - DART 재무는 보고서 종류별 표준 공시일 추정으로 시점 필터 + - 모든 데이터 조회를 as_of 기준 윈도우로 변경 + """ + backfill_mode = as_of is not None + today = as_of or date.today() week_ago = today - timedelta(days=7) + logger.info("scoring.start", as_of=str(today), backfill=backfill_mode) strong_buy: list = [] strong_sell: list = [] - # H3: KOSPI 일봉 갱신 후 시장 레짐 계산 - await fetch_kospi_ohlcv() + # H3: KOSPI 일봉 갱신 후 시장 레짐 계산 — 백필 모드는 이미 있는 데이터 사용, 갱신 스킵 + if not backfill_mode: + await fetch_kospi_ohlcv() - # 공식별 학습 가중치 로드 (없으면 균등 1.0) - formula_weights = {"magic": 1.0, "fscore": 1.0, "altman": 1.0, - "peg": 1.0, "momentum": 1.0, "beneish": 1.0, - "graph": 1.0} - async with pg_pool.acquire() as conn: - cfg = await conn.fetchrow( - "SELECT weights FROM weight_config ORDER BY config_date DESC LIMIT 1") - if cfg and cfg["weights"]: - try: - w = cfg["weights"] - if isinstance(w, str): - w = json.loads(w) - for k in formula_weights: - if k in w: - formula_weights[k] = float(w[k]) - except Exception as e: - logger.warning("weights.load_err", error=str(e)) + formula_weights = {f: 1.0 for f in ENSEMBLE_FORMULAS} + formula_weights["graph"] = 1.0 async with pg_pool.acquire() as conn: + # 사후 학습된 reliability/credibility 캐시 로드 — 백필 모드는 미적용 (look-ahead bias 차단) + await _load_reliability_caches(conn, backfill_mode=backfill_mode) + logger.info("reliability.cache.loaded", + reliability=len(_RELIABILITY_CACHE), source_cred=len(_SRC_CRED_CACHE)) # D: 시장 sentiment baseline 1회 계산 (전 종목 sentiment_alpha 산출용) market_sentiment_baseline = await calc_market_sentiment_baseline(conn, week_ago) logger.info("sentiment.market_baseline", value=round(market_sentiment_baseline, 2)) # H3: 시장 레짐 1회 계산 (전 종목 동일 적용) - regime_label, regime_adj = await calc_market_regime(conn) + regime_label, regime_adj = await calc_market_regime(conn, as_of=today if backfill_mode else None) + + # 공식별 학습 가중치 로드 — 현재 regime 매칭 우선, 없으면 segment='all' fallback + # 백필 모드는 config_date <= as_of 필터 (그 시점 이전 학습본만 적용) + seg_priority = [f"regime:{regime_label}", "all"] + for seg in seg_priority: + if backfill_mode: + cfg = await conn.fetchrow( + "SELECT weights, sample_size FROM weight_config " + "WHERE segment=$1 AND config_date <= $2 " + "ORDER BY config_date DESC LIMIT 1", seg, today) + else: + cfg = await conn.fetchrow( + "SELECT weights, sample_size FROM weight_config " + "WHERE segment=$1 ORDER BY config_date DESC LIMIT 1", seg) + if cfg and cfg["weights"]: + try: + w = cfg["weights"] + if isinstance(w, str): w = json.loads(w) + for k in formula_weights: + if k in w: formula_weights[k] = float(w[k]) + logger.info("weights.loaded", segment=seg, sample=cfg["sample_size"]) + break + except Exception as e: + logger.warning("weights.load_err", segment=seg, error=str(e)) await conn.execute(""" INSERT INTO market_regime (dt, regime, regime_adj) VALUES ($1, $2, $3) @@ -1542,10 +1699,10 @@ async def calculate_daily_scores(): flow_map: dict = {} macro_state: dict = {} try: - insider_map = await _load_insider_map(conn) - consensus_map = await _load_consensus_map(conn) - flow_map = await _load_inst_flow_map(conn) - macro_state = await _load_macro_state(conn) + insider_map = await _load_insider_map(conn, as_of=today if backfill_mode else None) + consensus_map = await _load_consensus_map(conn, as_of=today if backfill_mode else None) + flow_map = await _load_inst_flow_map(conn, as_of=today if backfill_mode else None) + macro_state = await _load_macro_state(conn, as_of=today if backfill_mode else None) logger.info("aux_signals.loaded", insider=len(insider_map), consensus=len(consensus_map), flow=len(flow_map), macro_inds=len(macro_state)) @@ -1642,11 +1799,12 @@ async def calculate_daily_scores(): dart_score = max(-100.0, min(100.0, (dart_pos - dart_neg) * 15)) # 가격/PER/PBR/시총 (Redis price:{code} → fallback: DB stock_prices) + # 백필 모드는 Redis·stock_prices 스킵하고 stock_ohlcv에서 그 시점 종가만 사용 price_score = 0.0 price_change = 0.0 has_price = False per = pbr = market_cap = 0.0 - if redis_cl: + if redis_cl and not backfill_mode: try: cached = await redis_cl.get(f"price:{stock}") if cached: @@ -1659,8 +1817,8 @@ async def calculate_daily_scores(): has_price = True except: pass - # DB fallback 1: stock_prices (장중 수집 데이터) - if not has_price: + # DB fallback 1: stock_prices (장중 수집 데이터) — 백필 모드 스킵 (30일 보존만) + if not has_price and not backfill_mode: try: pr = await conn.fetchrow( "SELECT change_pct, per, pbr, market_cap FROM stock_prices WHERE stock_code=$1 ORDER BY collected_at DESC LIMIT 1", @@ -1675,11 +1833,18 @@ async def calculate_daily_scores(): except: pass # DB fallback 2: stock_ohlcv 최근 종가 (장마감 후 price:{code} TTL 만료 시) + # 백필 모드: as_of 이전 종가만 사용. PER/PBR/시총은 0 (그 시점 데이터 없음). if not has_price: try: - ov = await conn.fetchrow( - "SELECT close_price, foreign_ratio FROM stock_ohlcv WHERE stock_code=$1 ORDER BY dt DESC LIMIT 1", - stock) + if backfill_mode: + ov = await conn.fetchrow( + "SELECT close_price, foreign_ratio FROM stock_ohlcv " + "WHERE stock_code=$1 AND dt <= $2 ORDER BY dt DESC LIMIT 1", + stock, today) + else: + ov = await conn.fetchrow( + "SELECT close_price, foreign_ratio FROM stock_ohlcv WHERE stock_code=$1 ORDER BY dt DESC LIMIT 1", + stock) if ov and ov["close_price"] > 0: # 전일 종가 기반, 당일 변동 미반영 → price_score=0 (중립) price_change = 0.0 @@ -1717,19 +1882,42 @@ async def calculate_daily_scores(): except: pass # 펀더멘털 점수 (dart_financials - 최신 사업보고서 기준) - fin_row = await conn.fetchrow(""" - SELECT f.roe, f.operating_margin, f.net_margin, f.debt_ratio, - f.fcf_ratio, f.revenue_growth, f.operating_profit, f.revenue, - f.operating_cashflow, f.total_equity, f.net_income, - f.total_assets, f.total_liabilities, - d.dps, d.dps_prev, d.dividend_yield - FROM dart_financials f - LEFT JOIN dart_dividends d ON d.stock_code = f.stock_code - AND d.bsns_year = f.bsns_year - WHERE f.stock_code = $1 - ORDER BY f.bsns_year DESC, f.reprt_code DESC - LIMIT 1 - """, stock) + # 백필 모드: 그 시점 이전 공시된 보고서만 사용 (estimated_disclosure_date <= as_of) + if backfill_mode: + fin_row = await conn.fetchrow(f""" + SELECT f.roe, f.operating_margin, f.net_margin, f.debt_ratio, + f.fcf_ratio, f.revenue_growth, f.operating_profit, f.revenue, + f.operating_cashflow, f.total_equity, f.net_income, + f.total_assets, f.total_liabilities, + d.dps, d.dps_prev, d.dividend_yield + FROM dart_financials f + LEFT JOIN dart_dividends d ON d.stock_code = f.stock_code + AND d.bsns_year = f.bsns_year + WHERE f.stock_code = $1 + AND (CASE f.reprt_code + WHEN '11011' THEN ((f.bsns_year::int + 1)::text || '-04-01')::date + WHEN '11012' THEN (f.bsns_year || '-05-16')::date + WHEN '11013' THEN (f.bsns_year || '-08-15')::date + WHEN '11014' THEN (f.bsns_year || '-11-15')::date + ELSE '9999-12-31'::date + END) <= $2 + ORDER BY f.bsns_year DESC, f.reprt_code DESC + LIMIT 1 + """, stock, today) + else: + fin_row = await conn.fetchrow(""" + SELECT f.roe, f.operating_margin, f.net_margin, f.debt_ratio, + f.fcf_ratio, f.revenue_growth, f.operating_profit, f.revenue, + f.operating_cashflow, f.total_equity, f.net_income, + f.total_assets, f.total_liabilities, + d.dps, d.dps_prev, d.dividend_yield + FROM dart_financials f + LEFT JOIN dart_dividends d ON d.stock_code = f.stock_code + AND d.bsns_year = f.bsns_year + WHERE f.stock_code = $1 + ORDER BY f.bsns_year DESC, f.reprt_code DESC + LIMIT 1 + """, stock) fin_data = dict(fin_row) if fin_row else {} # LEFT JOIN으로 NULL 가능 → 숫자 컬럼 일괄 정규화 (None → 0) for k in ("roe", "operating_margin", "net_margin", "debt_ratio", @@ -1742,14 +1930,25 @@ async def calculate_daily_scores(): # 최근 2년 연간 사업보고서 (F-Score year-over-year + 매직 포뮬러용) # 분기 보고서가 아닌 연간(11011)을 써야 영업이익/총자산 단위가 일치 - f_score_rows = await conn.fetch(""" - SELECT bsns_year, operating_margin, debt_ratio, revenue, - operating_cashflow, net_income, operating_profit, - total_assets, total_liabilities - FROM dart_financials - WHERE stock_code=$1 AND reprt_code='11011' - ORDER BY bsns_year DESC LIMIT 2 - """, stock) + if backfill_mode: + f_score_rows = await conn.fetch(""" + SELECT bsns_year, operating_margin, debt_ratio, revenue, + operating_cashflow, net_income, operating_profit, + total_assets, total_liabilities + FROM dart_financials + WHERE stock_code=$1 AND reprt_code='11011' + AND ((bsns_year::int + 1)::text || '-04-01')::date <= $2 + ORDER BY bsns_year DESC LIMIT 2 + """, stock, today) + else: + f_score_rows = await conn.fetch(""" + SELECT bsns_year, operating_margin, debt_ratio, revenue, + operating_cashflow, net_income, operating_profit, + total_assets, total_liabilities + FROM dart_financials + WHERE stock_code=$1 AND reprt_code='11011' + ORDER BY bsns_year DESC LIMIT 2 + """, stock) fin_curr = dict(f_score_rows[0]) if len(f_score_rows) >= 1 else {} fin_prev = dict(f_score_rows[1]) if len(f_score_rows) >= 2 else {} @@ -1762,7 +1961,7 @@ async def calculate_daily_scores(): fundamental_score, fin_reasons = calc_fundamental_score(fin_data, per, pbr) # H1: 5년 추세 점수 - trend_score, trend_reason = await calc_trend_score(conn, stock) + trend_score, trend_reason = await calc_trend_score(conn, stock, as_of=today if backfill_mode else None) if trend_reason: fin_reasons.append(trend_reason) @@ -1799,7 +1998,7 @@ async def calculate_daily_scores(): fin_reasons.append(peg_reason) # 12-1개월 모멘텀 (AQR) - mom_val, mom_sig, mom_reason = await calc_momentum(conn, stock) + mom_val, mom_sig, mom_reason = await calc_momentum(conn, stock, as_of=today if backfill_mode else None) if mom_reason: fin_reasons.append(mom_reason) @@ -1838,12 +2037,12 @@ async def calculate_daily_scores(): fin_reasons.append(g_reason) # Amihud 비유동성 (2002) — 소형 알파 - amihud_val, amihud_sig, amihud_reason = await calc_amihud(conn, stock) + amihud_val, amihud_sig, amihud_reason = await calc_amihud(conn, stock, as_of=today if backfill_mode else None) if amihud_reason: fin_reasons.append(amihud_reason) # 시장 베타 (BAB — Frazzini-Pedersen 2014) - beta_val, beta_sig, beta_reason = await calc_beta(conn, stock) + beta_val, beta_sig, beta_reason = await calc_beta(conn, stock, as_of=today if backfill_mode else None) if beta_reason: fin_reasons.append(beta_reason) @@ -2097,7 +2296,17 @@ async def calculate_daily_scores(): r["news_score"], r["dart_score"], r["price_score"], r["technical_score"], r["top_reasons"]) entry_price = 0 - if redis_cl: + # 백필 모드: Redis 스킵, stock_ohlcv에서 그 시점 종가 사용 (look-ahead bias 차단) + if backfill_mode: + try: + price_row = await conn.fetchrow( + "SELECT close_price FROM stock_ohlcv " + "WHERE stock_code=$1 AND dt<=$2 ORDER BY dt DESC LIMIT 1", + r["stock_code"], today) + if price_row and price_row["close_price"]: + entry_price = int(price_row["close_price"]) + except: pass + elif redis_cl: try: p_raw = await redis_cl.get(f"price:{r['stock_code']}") if p_raw: @@ -2380,6 +2589,84 @@ async def manual_calc(): return {"status": "done", "scored": n} +_BACKFILL_STATE: dict = {"running": False, "current": None, "done_days": 0, + "total_days": 0, "started_at": None, "errors": []} + + +@app.post("/score/backfill") +async def score_backfill(start_date: str = Query(...), end_date: str = Query(...), + skip_existing: bool = Query(default=True), + force: bool = Query(default=False)): + """과거 시점 score 백필 — look-ahead bias 차단 모드로 calculate_daily_scores 반복 호출. + 영업일(월~금)만 순회. skip_existing=True면 이미 stock_scores에 있는 날짜 건너뜀. + force=True면 동시 실행 잠금 무시 (위험). + + 예: POST /score/backfill?start_date=2025-06-01&end_date=2026-04-30 + 백그라운드 실행, 진행률은 GET /score/backfill/status 로 확인. + """ + if _BACKFILL_STATE["running"] and not force: + return {"status": "already_running", "state": _BACKFILL_STATE} + try: + s = datetime.strptime(start_date, "%Y-%m-%d").date() + e = datetime.strptime(end_date, "%Y-%m-%d").date() + except ValueError: + return {"status": "error", "msg": "start_date/end_date 형식 YYYY-MM-DD"} + if s > e: + return {"status": "error", "msg": "start_date > end_date"} + if e >= date.today(): + return {"status": "error", "msg": "end_date는 오늘 이전이어야 함 (오늘은 운영 score가 처리)"} + + # 영업일만 (월~금, 한국 공휴일 무시 — score는 휴일 데이터 자연스럽게 비어있음) + days: list[date] = [] + d = s + while d <= e: + if d.weekday() < 5: # 월=0 ~ 금=4 + days.append(d) + d += timedelta(days=1) + + if skip_existing: + async with pg_pool.acquire() as conn: + existing = await conn.fetch( + "SELECT DISTINCT score_date FROM stock_scores " + "WHERE score_date BETWEEN $1 AND $2", s, e) + existing_set = {r["score_date"] for r in existing} + days = [d for d in days if d not in existing_set] + + if not days: + return {"status": "nothing_to_do", "msg": "백필할 영업일 없음 (이미 score 있음)"} + + _BACKFILL_STATE.update({ + "running": True, "current": None, "done_days": 0, + "total_days": len(days), "started_at": datetime.now().isoformat(), + "errors": [], "range": f"{s} ~ {e}", + }) + + async def run(): + try: + for d in days: + _BACKFILL_STATE["current"] = str(d) + try: + await calculate_daily_scores(as_of=d) + except Exception as ex: + _BACKFILL_STATE["errors"].append({"date": str(d), "error": str(ex)[:200]}) + logger.error("score.backfill.day_err", date=str(d), error=str(ex)) + _BACKFILL_STATE["done_days"] += 1 + logger.info("score.backfill.done", days=len(days), + errors=len(_BACKFILL_STATE["errors"])) + finally: + _BACKFILL_STATE["running"] = False + _BACKFILL_STATE["current"] = None + + asyncio.create_task(run()) + return {"status": "started", "days": len(days), "from": str(days[0]), "to": str(days[-1])} + + +@app.get("/score/backfill/status") +async def score_backfill_status(): + """백필 진행 상태 + 에러 요약.""" + return _BACKFILL_STATE + + @app.post("/ohlcv/backfill") async def ohlcv_backfill(count: int = Query(default=0, ge=0), days: int = Query(default=400, ge=30, le=1200)): @@ -2760,167 +3047,596 @@ async def macro_kr(): return {"status": "ok", "data": out, "ts": datetime.now().isoformat()} +# ── 학습 보강: walk-forward CV + 평가지표 + 레짐/섹터 분리 ────────────── +LEARN_FEATURE_NAMES = [ + # 종합·앙상블 점수 + "total_score", "magic_score", "f_score", "altman_z", "peg", + "momentum_pct", "beneish_score", "gpa_pct", "g_score", + "amihud_illiq", "market_beta", + # 채널별 점수 + "news_score", "dart_score", "technical_score", + "foreign_score", "short_score", "price_score", + "us_overnight_adj", + # 펀더멘털·DCF·이익품질 + "trend_score", "earnings_quality", "margin_of_safety", + # 감성·뉴스 모멘텀 + "sentiment_momentum", "sentiment_alpha", + "attention_score", "news_surge_ratio", + # 변동성·레짐 + "volatility_60d", "market_regime_adj", +] + +def _row_features(r: dict) -> list[float]: + out = [] + for fn in LEARN_FEATURE_NAMES: + try: + out.append(float(r[fn] if r[fn] is not None else 0)) + except Exception: + out.append(0.0) + return out + + +async def _fetch_training_rows(conn, since: date, segment: str = "all"): + """ + 학습용 데이터: stock_scores 전체 이력 × stock_ohlcv 실현수익률. + (구버전은 recommendation_performance=추천종목만 → 표본 수십건·선택편향. + 현재는 전 종목 단면으로 확장 — 표본 100배+, 편향 제거.) + 수익률 = (score_date+N 거래일 종가 − score_date 종가) / score_date 종가 + · 7d exit = [+6,+11]일 윈도 첫 거래일 + · 30d exit = [+28,+38]일 윈도 첫 거래일 + 미래 미도달 종가는 NULL → 해당 horizon 학습에서 자동 제외(lookahead 차단) + segment: "all" | "regime:강세|중립|약세" | "sector:반도체..." + """ + where = "s.score_date >= $1" + params: list = [since] + if segment.startswith("regime:"): + params.append(segment.split(":", 1)[1]) + where += (f" AND EXISTS (SELECT 1 FROM market_regime mr " + f"WHERE mr.dt=s.score_date AND mr.regime=${len(params)})") + elif segment.startswith("sector:"): + params.append(segment.split(":", 1)[1]) + where += f" AND s.sector = ${len(params)}" + elif segment != "all": + return [] + feat_cols = ", ".join(f"s.{f}" for f in LEARN_FEATURE_NAMES) + rows = await conn.fetch(f""" + WITH t AS ( + SELECT s.stock_code, s.score_date, s.sector, s.signals, {feat_cols}, + (SELECT o.close_price FROM stock_ohlcv o + WHERE o.stock_code=s.stock_code + AND o.dt<=s.score_date AND o.dt>=s.score_date-7 + ORDER BY o.dt DESC LIMIT 1) AS entry_close, + (SELECT o.close_price FROM stock_ohlcv o + WHERE o.stock_code=s.stock_code + AND o.dt>=s.score_date+6 AND o.dt<=s.score_date+11 + ORDER BY o.dt ASC LIMIT 1) AS close_7d, + (SELECT o.close_price FROM stock_ohlcv o + WHERE o.stock_code=s.stock_code + AND o.dt>=s.score_date+28 AND o.dt<=s.score_date+38 + ORDER BY o.dt ASC LIMIT 1) AS close_30d, + (SELECT k.close_price FROM stock_ohlcv k + WHERE k.stock_code='KOSPI' + AND k.dt<=s.score_date AND k.dt>=s.score_date-7 + ORDER BY k.dt DESC LIMIT 1) AS kospi_entry, + (SELECT k.close_price FROM stock_ohlcv k + WHERE k.stock_code='KOSPI' + AND k.dt>=s.score_date+28 AND k.dt<=s.score_date+38 + ORDER BY k.dt ASC LIMIT 1) AS kospi_30d + FROM stock_scores s + WHERE {where} + ) + SELECT *, + CASE WHEN entry_close>0 AND close_7d IS NOT NULL + THEN (close_7d-entry_close)/entry_close*100 END AS return_7d, + CASE WHEN entry_close>0 AND close_30d IS NOT NULL + THEN (close_30d-entry_close)/entry_close*100 END AS return_30d, + CASE WHEN entry_close>0 AND close_30d IS NOT NULL + AND kospi_entry>0 AND kospi_30d IS NOT NULL + THEN (close_30d-entry_close)/entry_close*100 + - (kospi_30d-kospi_entry)/kospi_entry*100 END AS alpha_30d + FROM t + WHERE entry_close > 0 + ORDER BY score_date ASC, stock_code ASC + """, *params) + return rows + + +def _eval_metrics(y_true, y_pred) -> dict: + """IC(Spearman/Pearson), Hit ratio, Top-decile spread, Sharpe proxy, R²(OOS), MAE.""" + import numpy as np + try: + from scipy import stats as _ss + from sklearn.metrics import r2_score, mean_absolute_error + except Exception: + return {} + y_true = np.asarray(y_true, dtype=float) + y_pred = np.asarray(y_pred, dtype=float) + if len(y_true) < 3: + return {} + out = {} + try: + out["ic_spearman"] = round(float(_ss.spearmanr(y_true, y_pred).correlation), 4) + except Exception: + out["ic_spearman"] = None + try: + out["ic_pearson"] = round(float(_ss.pearsonr(y_true, y_pred)[0]), 4) + except Exception: + out["ic_pearson"] = None + try: + out["hit_ratio"] = round(float(((y_pred > 0) == (y_true > 0)).mean()), 4) + except Exception: + out["hit_ratio"] = None + # Top-decile spread (예측 상위 10% 실제 평균 - 하위 10% 실제 평균) + n = len(y_pred) + if n >= 10: + order = np.argsort(y_pred) + d = max(1, n // 10) + out["top_decile_spread"] = round(float(y_true[order[-d:]].mean() - y_true[order[:d]].mean()), 4) + else: + out["top_decile_spread"] = None + # Sharpe proxy: 예측 부호로 long/short했을 때 평균/표준편차 + try: + pnl = np.sign(y_pred) * y_true + sd = float(pnl.std()) + out["sharpe_proxy"] = round(float(pnl.mean() / sd * (252 ** 0.5)), 4) if sd > 0 else None + except Exception: + out["sharpe_proxy"] = None + try: + out["r2_oos"] = round(float(r2_score(y_true, y_pred)), 4) + except Exception: + out["r2_oos"] = None + try: + out["mae"] = round(float(mean_absolute_error(y_true, y_pred)), 4) + except Exception: + out["mae"] = None + return out + + +def _walk_forward_folds(rows, n_folds: int): + """ + 시간순 정렬된 rows를 (n_folds+1) 블록으로 나누어 expanding window 폴드 생성. + 각 fold i ∈ [1..n_folds]: train = [0..i*block], test = [i*block..(i+1)*block]. + test 데이터는 학습에 절대 안 들어감 → leakage 없음. + """ + n = len(rows) + if n < (n_folds + 1) * 3: + return [] + block = n // (n_folds + 1) + folds = [] + for i in range(1, n_folds + 1): + tr_end = i * block + te_end = min(n, (i + 1) * block) + tr = rows[:tr_end]; te = rows[tr_end:te_end] + if len(tr) < 5 or len(te) < 3: continue + folds.append((tr, te)) + return folds + + +def _aggregate_fold_metrics(metric_list): + """폴드별 metric dict의 평균. None은 제외.""" + if not metric_list: return {} + keys = set() + for m in metric_list: keys.update(m.keys()) + out = {} + for k in keys: + vals = [m[k] for m in metric_list if m.get(k) is not None] + out[k] = round(sum(vals) / len(vals), 4) if vals else None + return out + + @app.post("/learn-pricing") -async def learn_pricing(days: int = Query(default=90, ge=14, le=365)): +async def learn_pricing(days: int = Query(default=180, ge=14, le=730), + segment: str = Query(default="all"), + target: str = Query(default="return_30d"), + n_folds: int = Query(default=5, ge=2, le=10)): """ - D + E: 백테스트 데이터로 두 모델 학습 - - D: 단순 선형회귀 (점수 → 30일 수익률 계수) - - E: Random Forest (다변수 입력 → 30일 수익률) - 표본 부족 시 graceful (default 모델 또는 None) + Walk-forward CV로 가격 모델 학습 + 평가지표 산출. + + 피처: 종합/앙상블/펀더멘털/기술/감성/뉴스/변동성 등 26개 (LEARN_FEATURE_NAMES) + 모델: Linear / Random Forest / XGBoost (각각 별도 저장) + Segment: "all" | "regime:강세|중립|약세" | "sector:반도체|2차전지|..." + Target: "return_7d" | "return_30d" | "alpha_30d" + Lookahead bias 차단: score_date == rec_date, entry_price > 0, 시간순 expanding-window CV. """ + if target not in ("return_7d", "return_30d", "alpha_30d"): + return {"err": f"target은 return_7d|return_30d|alpha_30d 중 하나여야 함 (받음: {target})", + "segment": segment, "target": target} since = date.today() - timedelta(days=days) async with pg_pool.acquire() as conn: - rows = await conn.fetch(""" - SELECT s.total_score, s.magic_score, s.f_score, s.altman_z, - s.peg, s.momentum_pct, s.beneish_score, - p.return_30d, p.return_7d - FROM stock_scores s - JOIN recommendation_performance p - ON s.stock_code=p.stock_code AND s.score_date=p.rec_date - WHERE p.rec_date >= $1 - """, since) + rows = await _fetch_training_rows(conn, since, segment) + # target NULL 제거 (해당 horizon 도달 안 한 표본 제외) + rows = [r for r in rows if r[target] is not None] - out = {"period_days": days, "sample": len(rows)} - if len(rows) < 10: - out["msg"] = f"표본 {len(rows)} 부족 (최소 10) — 추천·성과 누적 후 재학습" - out["linear_coef"] = None - out["rf_feature_importance"] = None + out = {"period_days": days, "sample": len(rows), "segment": segment, "target": target} + if len(rows) < (n_folds + 1) * 3: + out["msg"] = (f"표본 {len(rows)} 부족 (walk-forward {n_folds}fold ≥ {(n_folds+1)*3} 필요) " + f"— 추천·성과 누적 후 재학습") return out try: import numpy as np - from sklearn.linear_model import LinearRegression + from sklearn.linear_model import Ridge from sklearn.ensemble import RandomForestRegressor - from sklearn.metrics import r2_score except Exception as e: return {**out, "err": f"sklearn import 실패: {e}"} - # D. 단순 선형회귀: total_score → return_30d (Walk-forward 적용) - valid_30d = [r for r in rows if r["return_30d"] is not None] - linear_summary = None - if len(valid_30d) >= 10: - X = np.array([[float(r["total_score"])] for r in valid_30d]) - y = np.array([float(r["return_30d"]) for r in valid_30d]) - m = LinearRegression().fit(X, y) - pred = m.predict(X) - # Walk-forward: 70/30 시간순 split (look-ahead bias 회피) - split = int(len(valid_30d) * 0.7) - oos_r2 = None - if split >= 5 and len(valid_30d) - split >= 3: - m_train = LinearRegression().fit(X[:split], y[:split]) - y_test_pred = m_train.predict(X[split:]) - oos_r2 = round(r2_score(y[split:], y_test_pred), 3) - linear_summary = { - "coef": round(float(m.coef_[0]), 4), - "intercept": round(float(m.intercept_), 4), - "r2_in_sample": round(r2_score(y, pred), 3), - "r2_out_of_sample_walkforward": oos_r2, - "n": len(valid_30d), - "interpretation": - f"점수 1점 상승 ≈ 30일 수익률 {m.coef_[0]:+.3f}%p, " - f"in-sample R²={r2_score(y, pred):.2f}, " - f"OOS R²={oos_r2 if oos_r2 is not None else 'N/A'} (look-ahead bias 회피)" - } + folds = _walk_forward_folds(rows, n_folds) + if not folds: + return {**out, "err": "fold 구성 실패 (표본 부족)"} - # E. Random Forest + XGBoost: 다변수 → return_30d - rf_summary = None - if len(valid_30d) >= 20: - feature_names = ["total_score", "magic_score", "f_score", "altman_z", - "peg", "momentum_pct", "beneish_score"] - X = np.array([[float(r[fn] or 0) for fn in feature_names] for r in valid_30d]) - y = np.array([float(r["return_30d"]) for r in valid_30d]) - rf = RandomForestRegressor(n_estimators=80, max_depth=5, random_state=42).fit(X, y) - importance = dict(zip(feature_names, [round(float(v), 3) for v in rf.feature_importances_])) - pred = rf.predict(X) - rf_summary = { - "n": len(valid_30d), - "r2_train": round(r2_score(y, pred), 3), - "feature_importance": dict(sorted(importance.items(), key=lambda x: -x[1])), - } - # XGBoost (gradient boosting) — RF보다 일반적으로 우월 - try: - import xgboost as xgb - xgb_model = xgb.XGBRegressor(n_estimators=100, max_depth=4, - learning_rate=0.05, random_state=42, - objective='reg:squarederror').fit(X, y) - xgb_pred = xgb_model.predict(X) - xgb_imp = dict(zip(feature_names, [round(float(v), 3) - for v in xgb_model.feature_importances_])) - rf_summary["xgb_r2"] = round(r2_score(y, xgb_pred), 3) - rf_summary["xgb_feature_importance"] = dict(sorted(xgb_imp.items(), key=lambda x: -x[1])) - except Exception as ex: - rf_summary["xgb_err"] = str(ex) - # 학습된 모델 직렬화 → DB 저장 (간단 버전: feature_importance만 JSONB로) - async with pg_pool.acquire() as conn: - await conn.execute(""" - CREATE TABLE IF NOT EXISTS pricing_model ( - model_date DATE PRIMARY KEY, - linear_coef FLOAT, linear_intercept FLOAT, linear_r2 FLOAT, - rf_features JSONB, rf_r2 FLOAT, sample_size INTEGER, - period_days INTEGER, created_at TIMESTAMP DEFAULT NOW() - ) - """) - await conn.execute(""" - INSERT INTO pricing_model (model_date, linear_coef, linear_intercept, - linear_r2, rf_features, rf_r2, sample_size, period_days) - VALUES (CURRENT_DATE, $1, $2, $3, $4, $5, $6, $7) - ON CONFLICT (model_date) DO UPDATE SET - linear_coef=$1, linear_intercept=$2, linear_r2=$3, - rf_features=$4, rf_r2=$5, sample_size=$6, period_days=$7 - """, linear_summary["coef"] if linear_summary else None, - linear_summary["intercept"] if linear_summary else None, - linear_summary["r2_in_sample"] if linear_summary else None, - json.dumps(importance), rf_summary["r2_train"], - len(valid_30d), days) + # ── 1. Linear (Ridge) ───────────────────────────────── + fold_metrics_lin = [] + last_lin = None + for tr, te in folds: + X_tr = np.array([_row_features(r) for r in tr]) + y_tr = np.array([float(r[target]) for r in tr]) + X_te = np.array([_row_features(r) for r in te]) + y_te = np.array([float(r[target]) for r in te]) + m = Ridge(alpha=1.0).fit(X_tr, y_tr) + y_p = m.predict(X_te) + fold_metrics_lin.append(_eval_metrics(y_te, y_p)) + last_lin = m # 마지막(최대 train) 모델 저장용 - return {**out, "linear": linear_summary, "rf": rf_summary, - "applied": "다음 /predict-price 호출부터 적용"} + linear_metrics = _aggregate_fold_metrics(fold_metrics_lin) + + # 전체 데이터 재학습 (배포용 모델) + X_all = np.array([_row_features(r) for r in rows]) + y_all = np.array([float(r[target]) for r in rows]) + final_lin = Ridge(alpha=1.0).fit(X_all, y_all) + lin_coef = {fn: round(float(c), 6) for fn, c in zip(LEARN_FEATURE_NAMES, final_lin.coef_)} + + # ── 2. Random Forest ────────────────────────────────── + fold_metrics_rf = [] + for tr, te in folds: + X_tr = np.array([_row_features(r) for r in tr]) + y_tr = np.array([float(r[target]) for r in tr]) + X_te = np.array([_row_features(r) for r in te]) + y_te = np.array([float(r[target]) for r in te]) + m = RandomForestRegressor(n_estimators=200, max_depth=6, + min_samples_leaf=3, random_state=42, n_jobs=2).fit(X_tr, y_tr) + fold_metrics_rf.append(_eval_metrics(y_te, m.predict(X_te))) + rf_metrics = _aggregate_fold_metrics(fold_metrics_rf) + final_rf = RandomForestRegressor(n_estimators=200, max_depth=6, + min_samples_leaf=3, random_state=42, n_jobs=2).fit(X_all, y_all) + rf_imp = dict(zip(LEARN_FEATURE_NAMES, + [round(float(v), 4) for v in final_rf.feature_importances_])) + rf_imp = dict(sorted(rf_imp.items(), key=lambda x: -x[1])) + + # ── 3. XGBoost ──────────────────────────────────────── + xgb_metrics = None; xgb_imp = None + try: + import xgboost as xgb + fold_metrics_xgb = [] + for tr, te in folds: + X_tr = np.array([_row_features(r) for r in tr]) + y_tr = np.array([float(r[target]) for r in tr]) + X_te = np.array([_row_features(r) for r in te]) + y_te = np.array([float(r[target]) for r in te]) + m = xgb.XGBRegressor(n_estimators=200, max_depth=4, learning_rate=0.05, + subsample=0.85, colsample_bytree=0.85, + random_state=42, objective='reg:squarederror', + n_jobs=2).fit(X_tr, y_tr) + fold_metrics_xgb.append(_eval_metrics(y_te, m.predict(X_te))) + xgb_metrics = _aggregate_fold_metrics(fold_metrics_xgb) + final_xgb = xgb.XGBRegressor(n_estimators=200, max_depth=4, learning_rate=0.05, + subsample=0.85, colsample_bytree=0.85, + random_state=42, objective='reg:squarederror', + n_jobs=2).fit(X_all, y_all) + xgb_imp = dict(zip(LEARN_FEATURE_NAMES, + [round(float(v), 4) for v in final_xgb.feature_importances_])) + xgb_imp = dict(sorted(xgb_imp.items(), key=lambda x: -x[1])) + except Exception as ex: + xgb_metrics = {"err": str(ex)} + + # ── 4. 저장 (pricing_model_v2 + model_metrics) ───────── + today_d = date.today() + async with pg_pool.acquire() as conn: + # Linear + await conn.execute(""" + INSERT INTO pricing_model_v2 + (model_date, segment, model_type, target, + feature_names, coef, intercept, + r2_oos, ic_spearman, hit_ratio, sample_size, period_days) + VALUES ($1,$2,'linear',$3,$4::jsonb,$5::jsonb,$6,$7,$8,$9,$10,$11) + ON CONFLICT (model_date, segment, model_type, target) DO UPDATE SET + feature_names=$4::jsonb, coef=$5::jsonb, intercept=$6, + r2_oos=$7, ic_spearman=$8, hit_ratio=$9, + sample_size=$10, period_days=$11 + """, today_d, segment, target, + json.dumps(LEARN_FEATURE_NAMES), json.dumps(lin_coef), + float(final_lin.intercept_), + linear_metrics.get("r2_oos"), linear_metrics.get("ic_spearman"), + linear_metrics.get("hit_ratio"), len(rows), days) + # RF + await conn.execute(""" + INSERT INTO pricing_model_v2 + (model_date, segment, model_type, target, + feature_names, feature_importance, model_blob, + r2_oos, ic_spearman, hit_ratio, sample_size, period_days) + VALUES ($1,$2,'rf',$3,$4::jsonb,$5::jsonb,$6,$7,$8,$9,$10,$11) + ON CONFLICT (model_date, segment, model_type, target) DO UPDATE SET + feature_names=$4::jsonb, feature_importance=$5::jsonb, model_blob=$6, + r2_oos=$7, ic_spearman=$8, hit_ratio=$9, + sample_size=$10, period_days=$11 + """, today_d, segment, target, + json.dumps(LEARN_FEATURE_NAMES), json.dumps(rf_imp), + pickle.dumps(final_rf), + rf_metrics.get("r2_oos"), rf_metrics.get("ic_spearman"), + rf_metrics.get("hit_ratio"), len(rows), days) + # XGBoost (성공 시) + if xgb_imp is not None: + await conn.execute(""" + INSERT INTO pricing_model_v2 + (model_date, segment, model_type, target, + feature_names, feature_importance, model_blob, + r2_oos, ic_spearman, hit_ratio, sample_size, period_days) + VALUES ($1,$2,'xgb',$3,$4::jsonb,$5::jsonb,$6,$7,$8,$9,$10,$11) + ON CONFLICT (model_date, segment, model_type, target) DO UPDATE SET + feature_names=$4::jsonb, feature_importance=$5::jsonb, model_blob=$6, + r2_oos=$7, ic_spearman=$8, hit_ratio=$9, + sample_size=$10, period_days=$11 + """, today_d, segment, target, + json.dumps(LEARN_FEATURE_NAMES), json.dumps(xgb_imp), + pickle.dumps(final_xgb), + xgb_metrics.get("r2_oos"), xgb_metrics.get("ic_spearman"), + xgb_metrics.get("hit_ratio"), len(rows), days) + # model_metrics — 폴드 평균값 저장 + for mtype, m in (("linear", linear_metrics), ("rf", rf_metrics), + ("xgb", xgb_metrics if xgb_imp is not None else None)): + if not m or "err" in m: continue + imp_dict = (lin_coef if mtype == "linear" + else rf_imp if mtype == "rf" else (xgb_imp or {})) + await conn.execute(""" + INSERT INTO model_metrics + (model_date, model_type, segment, target, + period_days, sample_size, n_folds, + ic_spearman, ic_pearson, hit_ratio, top_decile_spread, + sharpe_proxy, r2_oos, mae, feature_importance) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15::jsonb) + """, today_d, mtype, segment, target, + days, len(rows), len(folds), + m.get("ic_spearman"), m.get("ic_pearson"), m.get("hit_ratio"), + m.get("top_decile_spread"), m.get("sharpe_proxy"), + m.get("r2_oos"), m.get("mae"), json.dumps(imp_dict)) + + return {**out, "n_folds": len(folds), "n_features": len(LEARN_FEATURE_NAMES), + "linear": linear_metrics, "rf": rf_metrics, "xgb": xgb_metrics, + "rf_top_importance": dict(list(rf_imp.items())[:8]), + "applied": "다음 /predict-price 호출부터 적용 (segment 일치 모델 우선)"} @app.get("/predict-price/{code}") -async def predict_price(code: str): - """학습된 모델로 N일 후 예상 수익률·가격 추정""" +async def predict_price(code: str, model_type: str = Query(default="rf"), + target: str = Query(default="return_30d")): + """ + 학습된 모델로 예상 수익률·가격 추정. + + 모델 선택 우선순위: + 1) 종목 sector × 현재 regime 일치 segment 모델 + 2) 현재 regime 일치 segment 모델 + 3) sector 일치 segment 모델 + 4) segment='all' 모델 (default) + """ async with pg_pool.acquire() as conn: - s = await conn.fetchrow(""" - SELECT total_score, magic_score, f_score, altman_z, - peg, momentum_pct, beneish_score, sector + s = await conn.fetchrow(f""" + SELECT {", ".join(LEARN_FEATURE_NAMES)}, sector FROM stock_scores WHERE stock_code=$1 ORDER BY score_date DESC LIMIT 1 """, code) - m = await conn.fetchrow(""" - SELECT linear_coef, linear_intercept, linear_r2, rf_r2, sample_size - FROM pricing_model ORDER BY model_date DESC LIMIT 1 - """) - # 현재가 — Redis or stock_prices - cur_price = 0 - if redis_cl: - try: - p = await redis_cl.get(f"price:{code}") - if p: - cur_price = int(json.loads(p).get("price") or 0) - except: pass - if not cur_price: - pr = await conn.fetchrow( - "SELECT price FROM stock_prices WHERE stock_code=$1 ORDER BY collected_at DESC LIMIT 1", - code) - if pr: cur_price = int(pr["price"] or 0) + if not s: + return {"code": code, "msg": "stock_scores 데이터 없음"} + # 현재 regime + rg_row = await conn.fetchrow( + "SELECT regime FROM market_regime ORDER BY dt DESC LIMIT 1") + cur_regime = (rg_row["regime"] if rg_row else "중립") or "중립" + sector = (s["sector"] or "").strip() + candidates = [] + if sector: + candidates.append(f"sector:{sector}") + if cur_regime: + candidates.append(f"regime:{cur_regime}") + candidates.append("all") + + m = None; m_seg = None + for seg in candidates: + m = await conn.fetchrow(""" + SELECT segment, model_type, feature_names, coef, intercept, + feature_importance, model_blob, + r2_oos, ic_spearman, hit_ratio, sample_size + FROM pricing_model_v2 + WHERE segment=$1 AND model_type=$2 AND target=$3 + ORDER BY model_date DESC LIMIT 1 + """, seg, model_type, target) + if m: + m_seg = seg + break + # 현재가 + cur_price = await get_current_price(code) - if not s: - return {"code": code, "msg": "stock_scores 데이터 없음"} if not m: return {"code": code, "msg": "학습 모델 없음 — 먼저 /learn-pricing 호출"} - pred_30d_pct = (m["linear_intercept"] or 0) + (m["linear_coef"] or 0) * float(s["total_score"]) - pred_price = int(cur_price * (1 + pred_30d_pct / 100)) if cur_price else None + # 예측: linear=coef 직접계산 / RF·XGB=저장된 모델(pickle) 로드 추론 + pred_pct = None + used_method = model_type + if model_type == "linear": + coef = m["coef"] + if isinstance(coef, str): + try: coef = json.loads(coef) + except: coef = {} + intercept = float(m["intercept"] or 0) + pred_pct = intercept + for fn in LEARN_FEATURE_NAMES: + pred_pct += float(coef.get(fn, 0) or 0) * float(s[fn] or 0) + else: + # RF/XGB: 직렬화 저장된 모델로 직접 추론 + blob = m["model_blob"] + if blob: + try: + import numpy as np + model = pickle.loads(blob) + fnames = m["feature_names"] + if isinstance(fnames, str): + try: fnames = json.loads(fnames) + except: fnames = [] + fnames = fnames or LEARN_FEATURE_NAMES + vals = [] + for fn in fnames: + try: vals.append(float(s[fn] if s[fn] is not None else 0)) + except Exception: vals.append(0.0) + pred_pct = float(model.predict(np.array([vals], dtype=float))[0]) + except Exception as e: + logger.warning("predict.blob_err", model_type=model_type, err=str(e)) + if pred_pct is None: + # 저장 모델 없음/실패 → linear 계수로 fallback + used_method = "linear(fallback)" + lm = await pg_pool.fetchrow(""" + SELECT coef, intercept FROM pricing_model_v2 + WHERE segment=$1 AND model_type='linear' AND target=$2 + ORDER BY model_date DESC LIMIT 1 + """, m_seg, target) + if lm: + coef = lm["coef"] + if isinstance(coef, str): + try: coef = json.loads(coef) + except: coef = {} + pred_pct = float(lm["intercept"] or 0) + for fn in LEARN_FEATURE_NAMES: + pred_pct += float(coef.get(fn, 0) or 0) * float(s[fn] or 0) + if pred_pct is None: + return {"code": code, "msg": f"{model_type} 모델 학습 불가 — linear fallback도 없음"} + + horizon_days = {"return_7d": 7, "return_30d": 30, "alpha_30d": 30}.get(target, 30) + pred_price = int(cur_price * (1 + pred_pct / 100)) if cur_price else None return { "code": code, "current_price": cur_price, - "current_score": float(s["total_score"]), - "predicted_30d_return_pct": round(pred_30d_pct, 2), - "predicted_30d_price": pred_price, - "model": {"r2": m["linear_r2"], "n": m["sample_size"]}, - "disclaimer": "선형 회귀 기반 단순 추정. 신뢰도 R² 참고. RF 모델은 feature 기반 더 정교 (내부)", + "current_regime": cur_regime, + "sector": sector, + "model_used": {"segment": m_seg, "type": used_method, "target": target, + "r2_oos": m["r2_oos"], "ic_spearman": m["ic_spearman"], + "hit_ratio": m["hit_ratio"], "n": m["sample_size"]}, + "predicted_return_pct": round(pred_pct, 2) if pred_pct is not None else None, + "horizon_days": horizon_days, + "predicted_price": pred_price, + "disclaimer": "Walk-forward CV 학습. IC > 0.05 면 유효 신호. RF/XGB는 저장된 모델로 직접 추론.", + } + + +@app.get("/model-metrics") +async def model_metrics_view(days: int = Query(default=30), segment: str = Query(default="")): + """최근 학습된 모델들의 walk-forward CV 평가지표 조회""" + since = date.today() - timedelta(days=days) + where = "model_date >= $1" + params: list = [since] + if segment: + params.append(segment) + where += f" AND segment = ${len(params)}" + async with pg_pool.acquire() as conn: + rows = await conn.fetch(f""" + SELECT model_date, model_type, segment, target, + sample_size, n_folds, + ic_spearman, ic_pearson, hit_ratio, + top_decile_spread, sharpe_proxy, r2_oos, mae + FROM model_metrics WHERE {where} + ORDER BY model_date DESC, segment ASC, model_type ASC + """, *params) + return {"count": len(rows), "rows": [dict(r) for r in rows]} + + +@app.get("/portfolio/recommended") +async def portfolio_recommended(amount: int = Query(default=0, ge=0), + max_stocks: int = Query(default=12, ge=3, le=30), + sector_cap: float = Query(default=30.0, ge=10, le=100)): + """오늘 추천 종목으로 분산 포트폴리오 구성. + 비중 = position_size_pct(변동성·점수 가중) 정규화 + 섹터 상한(기본 30%). + amount(원)를 주면 종목별 배분금액·매수가능주수까지 계산.""" + async with pg_pool.acquire() as conn: + rows = await conn.fetch(""" + SELECT s.stock_code, s.stock_name, s.total_score, s.recommendation, + COALESCE(s.position_size_pct, 0) AS psize, + COALESCE(NULLIF(s.sector, ''), '기타') AS sector, + s.top_reasons + FROM stock_scores s + JOIN dart_corps d ON d.stock_code = s.stock_code AND d.is_active = true + WHERE s.score_date = (SELECT MAX(score_date) FROM stock_scores) + AND s.recommendation IN ('강력매수', '매수관심') + ORDER BY s.total_score DESC + LIMIT $1 + """, max_stocks) + if not rows: + return {"as_of": str(date.today()), "n_stocks": 0, "holdings": [], + "sector_breakdown": {}, + "msg": "현재 추천(강력매수·매수관심) 종목이 없습니다."} + + def _reason(tr): + if isinstance(tr, list): + return " · ".join(str(x) for x in tr[:2])[:80] + return str(tr)[:80] if tr else "" + + items = [] + for r in rows: + psize = float(r["psize"] or 0) + raw = psize if psize > 0 else max(1.0, float(r["total_score"]) / 12.0) + items.append({ + "stock_code": r["stock_code"], + "stock_name": r["stock_name"] or r["stock_code"], + "total_score": round(float(r["total_score"]), 1), + "recommendation": r["recommendation"], + "sector": r["sector"], "raw": raw, "weight": 0.0, + "reason": _reason(r["top_reasons"]), + }) + + def _normalize(its): + tot = sum(i["raw"] for i in its) or 1.0 + for i in its: + i["weight"] = i["raw"] / tot * 100.0 + + _normalize(items) + # 섹터 상한 반복 적용 — 한 섹터가 sector_cap 초과 시 축소 후 재정규화 + for _ in range(4): + ssum: dict = {} + for i in items: + ssum[i["sector"]] = ssum.get(i["sector"], 0.0) + i["weight"] + over = {s: v for s, v in ssum.items() if v > sector_cap + 0.01} + if not over: + break + for i in items: + i["raw"] = (i["weight"] * sector_cap / ssum[i["sector"]] + if i["sector"] in over else i["weight"]) + _normalize(items) + + holdings = [] + for i in items: + price = await get_current_price(i["stock_code"]) if amount > 0 else 0 + alloc = int(amount * i["weight"] / 100.0) if amount > 0 else 0 + shares = (alloc // price) if (amount > 0 and price > 0) else 0 + holdings.append({ + "stock_code": i["stock_code"], "stock_name": i["stock_name"], + "sector": i["sector"], "total_score": i["total_score"], + "recommendation": i["recommendation"], + "weight_pct": round(i["weight"], 2), + "price": price, "alloc_amount": alloc, "shares": shares, + "invest_amount": shares * price, "reason": i["reason"], + }) + holdings.sort(key=lambda h: -h["weight_pct"]) + + sector_breakdown: dict = {} + for h in holdings: + sector_breakdown[h["sector"]] = round( + sector_breakdown.get(h["sector"], 0.0) + h["weight_pct"], 2) + sector_breakdown = dict(sorted(sector_breakdown.items(), key=lambda x: -x[1])) + + invested = sum(h["invest_amount"] for h in holdings) + return { + "as_of": str(date.today()), + "n_stocks": len(holdings), + "input_amount": amount, + "invested_amount": invested, + "cash_remaining": max(0, amount - invested) if amount > 0 else 0, + "sector_breakdown": sector_breakdown, + "holdings": holdings, }