#!/usr/bin/env python3 """ build_capital_style_allocation_v1.py ─────────────────────────────────────────────────────────────────────────────── CAPITAL_STYLE_ALLOCATION_V1 — 투자성향별 자금 유동성 가중 엔진 4개 투자성향(SCALP/SWING/MOMENTUM/POSITION)에 대해 기존 신호 빌딩블록을 성향별 다른 가중치로 융합하여 결정론적 conviction_score(0~100)와 recommended_position_pct를 산출한다. 신호 출처 (LLM 계산 0%): - technical_score : data_feed RSI14/Disparity/Ret5D 기반 규칙 계산 - smart_money_score : Temp/smart_money_flow_signal_v2.json (이미 0~100) - fundamental_score : Temp/fundamental_multifactor_v3.json (이미 0~100) - macro_event_score : Temp/macro_event_ticker_impact_v1.json ([-100,+100]→[0,100]) - liquidity_modifier : Temp/liquidity_flow_signal_v1.json (DEEP/MODERATE/THIN/FROZEN→배수) 가중치 출처: EXPERT_PRIOR (spec/calibration_registry.yaml 등록) SCALP: tech=0.50, smart=0.30, fund=0.05, macro=0.15 SWING: tech=0.30, smart=0.35, fund=0.15, macro=0.20 MOMENTUM: tech=0.15, smart=0.25, fund=0.40, macro=0.20 POSITION: tech=0.10, smart=0.20, fund=0.55, macro=0.15 출력: Temp/capital_style_allocation_v1.json 사용법: python tools/build_capital_style_allocation_v1.py """ from __future__ import annotations import argparse import json import math import sys from pathlib import Path ROOT = Path(__file__).resolve().parent.parent if sys.stdout.encoding and sys.stdout.encoding.lower() not in ("utf-8", "utf8"): sys.stdout = open(sys.stdout.fileno(), mode="w", encoding="utf-8", buffering=1) # ─── 가중치 정의 (EXPERT_PRIOR — calibration_registry.yaml 등록) ─────────── W_STYLE: dict[str, dict[str, float]] = { "SCALP": {"technical": 0.50, "smartmoney": 0.30, "fundamental": 0.05, "macro_event": 0.15}, "SWING": {"technical": 0.30, "smartmoney": 0.35, "fundamental": 0.15, "macro_event": 0.20}, "MOMENTUM": {"technical": 0.15, "smartmoney": 0.25, "fundamental": 0.40, "macro_event": 0.20}, "POSITION": {"technical": 0.10, "smartmoney": 0.20, "fundamental": 0.55, "macro_event": 0.15}, } # ─── 포지션 사이즈 맵핑 (EXPERT_PRIOR) ──────────────────────────────────── def conviction_to_pct(score: float) -> float: if score >= 80: return 7.0 if score >= 65: return 5.0 if score >= 50: return 3.0 if score >= 35: return 1.5 return 0.0 # ─── 유동성 배수 ────────────────────────────────────────────────────────── LIQUIDITY_MODIFIER: dict[str, float] = { "DEEP": 1.00, "MODERATE": 0.90, "THIN": 0.75, "FROZEN": 0.00, } def _load(path: Path) -> dict | list: if not path.exists(): return {} try: d = json.loads(path.read_text(encoding="utf-8")) return d except Exception: return {} def _rows(v: object) -> list[dict]: if isinstance(v, list): return [x for x in v if isinstance(x, dict)] if isinstance(v, dict): for key in ("rows", "data", "tickers"): c = v.get(key) if isinstance(c, list): return [x for x in c if isinstance(x, dict)] return [] def _f(v: object, default: float = 0.0) -> float: try: return float(v) # type: ignore[arg-type] except Exception: return default def _clamp(v: float, lo: float = 0.0, hi: float = 100.0) -> float: return max(lo, min(hi, v)) # ─── 기술적 점수 산출 (결정론적 규칙) ──────────────────────────────────── def compute_technical_score(rsi14: float, disparity: float, ret5d: float, volume: float, avg_vol5d: float) -> float: """ 기본 50점 기준, 아래 조건을 가산/감산. RSI14 < 35 → +20 (과매도 반등 기회) RSI14 > 70 → -25 (과매수 추격 위험) Disparity < 3% → +15 (MA20 근접 — 눌림목) Disparity > 10% → -20 (MA20 과이격) Ret5D < -5% → +10 (단기 급락 반등 후보) 거래량 확인(volume >= avgVol5d*1.2 AND Ret5D > 0) → +10 (수급 확인 돌파) """ score = 50.0 if rsi14 < 35: score += 20.0 elif rsi14 > 70: score -= 25.0 if disparity < 3.0: score += 15.0 elif disparity > 10.0: score -= 20.0 if ret5d < -5.0: score += 10.0 if avg_vol5d > 0 and volume >= avg_vol5d * 1.2 and ret5d > 0: score += 10.0 return _clamp(score) # ─── 메인 ───────────────────────────────────────────────────────────────── def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--json", default=str(ROOT / "GatherTradingData.json")) ap.add_argument("--smart", default=str(ROOT / "Temp/smart_money_flow_signal_v2.json")) ap.add_argument("--fund", default=str(ROOT / "Temp/fundamental_multifactor_v3.json")) ap.add_argument("--macro", default=str(ROOT / "Temp/macro_event_ticker_impact_v1.json")) ap.add_argument("--liquidity",default=str(ROOT / "Temp/liquidity_flow_signal_v1.json")) ap.add_argument("--out", default=str(ROOT / "Temp/capital_style_allocation_v1.json")) args = ap.parse_args() def rp(s: str) -> Path: p = Path(s) return p if p.is_absolute() else ROOT / p # ── 입력 로드 ──────────────────────────────────────────────────────── payload = _load(rp(args.json)) smart_raw = _load(rp(args.smart)) fund_raw = _load(rp(args.fund)) macro_raw = _load(rp(args.macro)) liq_raw = _load(rp(args.liquidity)) data = payload.get("data", {}) if isinstance(payload, dict) else {} df_list = _rows(data.get("data_feed")) if isinstance(data, dict) else [] # ── 신호 인덱스 구성 ────────────────────────────────────────────────── smart_map: dict[str, float] = {} for r in _rows(smart_raw): t = str(r.get("ticker") or "") smart_map[t] = _f(r.get("smart_money_score"), 50.0) fund_map: dict[str, float] = {} for r in _rows(fund_raw): t = str(r.get("ticker") or "") fund_map[t] = _f(r.get("score"), 50.0) macro_map: dict[str, dict] = {} macro_tickers = _rows(macro_raw) if isinstance(macro_raw, list) else _rows(macro_raw.get("tickers") if isinstance(macro_raw, dict) else []) for r in macro_tickers: t = str(r.get("ticker") or "") macro_map[t] = r liq_map: dict[str, str] = {} exec_map: dict[str, str] = {} for r in _rows(liq_raw): t = str(r.get("ticker") or "") liq_map[t] = str(r.get("liquidity_label") or "MODERATE") exec_map[t] = str(r.get("execution_mode") or "") # ── 종목별 신호 산출 + 성향별 conviction ─────────────────────────── rows_out: list[dict] = [] errors: list[str] = [] for r in df_list: ticker = str(r.get("Ticker") or r.get("ticker") or "") name = str(r.get("Name") or r.get("name") or "") if not ticker: continue # 기술지표 rsi14 = _f(r.get("RSI14"), 50.0) disparity = _f(r.get("Disparity"), 0.0) ret5d = _f(r.get("Ret5D"), 0.0) volume = _f(r.get("Volume"), 0.0) avg5d = _f(r.get("AvgVolume_5D"),0.0) # 4개 신호 정규화 [0, 100] tech_score = compute_technical_score(rsi14, disparity, ret5d, volume, avg5d) smart_score = _clamp(smart_map.get(ticker, 50.0)) fund_score = _clamp(fund_map.get(ticker, 50.0)) macro_info = macro_map.get(ticker, {}) macro_impact = _f(macro_info.get("primary_impact_score"), 0.0) macro_gate = str(macro_info.get("primary_gate") or "NEUTRAL") if macro_gate == "AVOID_NEW_BUY": macro_score = 0.0 else: macro_score = _clamp((macro_impact + 100.0) / 2.0) # [-100,+100]→[0,100] # 유동성 배수 liq_label = liq_map.get(ticker, "MODERATE") exec_mode = exec_map.get(ticker, "") if exec_mode == "FROZEN" or liq_label == "FROZEN": liq_modifier = 0.0 else: liq_modifier = LIQUIDITY_MODIFIER.get(liq_label, 0.90) signal_breakdown = { "technical_score": round(tech_score, 2), "smart_money_score": round(smart_score, 2), "fundamental_score": round(fund_score, 2), "macro_event_score": round(macro_score, 2), "macro_gate": macro_gate, "liquidity_label": liq_label, "liquidity_modifier": liq_modifier, } # 4개 성향별 conviction 산출 style_rows: list[dict] = [] for style, weights in W_STYLE.items(): raw = (weights["technical"] * tech_score + weights["smartmoney"] * smart_score + weights["fundamental"] * fund_score + weights["macro_event"] * macro_score) conviction = _clamp(round(raw * liq_modifier, 2)) rec_pct = conviction_to_pct(conviction) # 범위 검증 if not (0.0 <= conviction <= 100.0): errors.append(f"{ticker}.{style}: conviction={conviction} out of [0,100]") if not (0.0 <= rec_pct <= 7.0): errors.append(f"{ticker}.{style}: recommended_pct={rec_pct} out of [0,7]") style_rows.append({ "style": style, "conviction_score": conviction, "recommended_pct": rec_pct, "raw_weighted_score": round(raw, 2), }) rows_out.append({ "ticker": ticker, "name": name, "signal_breakdown": signal_breakdown, "styles": style_rows, "formula_id": "CAPITAL_STYLE_ALLOCATION_V1", }) gate = "PASS" if not errors and rows_out else ("FAIL" if errors else "NO_DATA") best_summary: dict[str, object] = {} if rows_out: ranked: list[tuple[float, float, dict[str, object], dict[str, object]]] = [] for row in rows_out: styles = [s for s in (row.get("styles") or []) if isinstance(s, dict)] if not styles: continue best_style = max(styles, key=lambda s: _f(s.get("conviction_score"), 0.0)) best_conviction = _f(best_style.get("conviction_score"), 0.0) best_pct = _f(best_style.get("recommended_pct"), 0.0) ranked.append((best_conviction, best_pct, row, best_style)) if ranked: ranked.sort(key=lambda item: (item[0], item[1], str(item[2].get("ticker") or "")), reverse=True) best_conviction, best_pct, best_row, best_style = ranked[0] best_summary = { "capital_style_conviction": round(best_conviction, 2), "capital_style_label": best_style.get("style") or "UNKNOWN", "capital_style_ticker": best_row.get("ticker") or "", "capital_style_name": best_row.get("name") or "", "capital_style_recommended_pct": round(best_pct, 2), } result = { "formula_id": "CAPITAL_STYLE_ALLOCATION_V1", "gate": gate, "ticker_count": len(rows_out), "style_list": list(W_STYLE.keys()), "weights": W_STYLE, "rows": rows_out, "errors": errors, **best_summary, "meta": { "weight_source": "EXPERT_PRIOR", "sample_n": 0, "llm_computed": False, "deterministic": True, "unvalidated_weight_label": "UNVALIDATED_WEIGHT", "calibration_note": "spec/calibration_registry.yaml 등록. " "실측 30건 누적 후 PROVISIONAL→CALIBRATED 승격 필요.", }, } out_path = rp(args.out) out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") sep = "=" * 70 print(sep) print(" CAPITAL_STYLE_ALLOCATION_V1") print(sep) print(f" gate={gate} tickers={len(rows_out)} errors={len(errors)}") for r in rows_out[:3]: best = max(r["styles"], key=lambda s: s["conviction_score"]) print(f" {r['ticker']:<10} {r['name'][:12]:<12} " f"best_style={best['style']:<10} conviction={best['conviction_score']:.1f} " f"rec_pct={best['recommended_pct']:.1f}%") if len(rows_out) > 3: print(f" ... 외 {len(rows_out)-3}개") if errors: print(f"\n [!] 오류 {len(errors)}건:") for e in errors[:5]: print(f" {e}") print(f"\n → 저장: {out_path}") print(f" {'CAPITAL_ALLOC_BUILD_OK' if gate=='PASS' else 'CAPITAL_ALLOC_BUILD_FAIL'}\n") return 0 if gate == "PASS" else 1 if __name__ == "__main__": raise SystemExit(main())