From 5c721e6b703d582a6f8555c4890b07053962d74d Mon Sep 17 00:00:00 2001 From: kyu Date: Wed, 3 Jun 2026 12:29:02 +0900 Subject: [PATCH] =?UTF-8?q?feat:=20=EB=8D=B0=EC=9D=B4=ED=84=B0=20=ED=86=A0?= =?UTF-8?q?=EB=8C=80=20+=20=EC=A0=95=ED=99=95=EB=8F=84=20=EA=B2=80?= =?UTF-8?q?=EC=A6=9D=20(AI=20=ED=88=AC=EC=9E=90=EC=A1=B0=EC=A7=81=20Phase?= =?UTF-8?q?=200~1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - score-engine: 일간리포트 텔레그램을 notify=True(16:30 1회)로 게이팅 → 호출마다 발신되던 폭주 제거. 데이터 무결성 모니터(/data-health + 평일 매시간 경고). 정확도 검증 하베스트(/accuracy + 주간 리포트) — 추천 등급별 실측 알파/적중률. - ta-engine: job_analyze가 is_active=true 전 활성종목 시총순 커버(장중 상위500· 장마감 전종목). 기존 LIMIT 500·무필터로 LS 등 누락되던 버그 수정. - docs/ai_org.md: 데이터우선 마스터 기획(데이터→검증→지능→실행). Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/ai_org.md | 68 +++++++++++++++++++++++ score-engine/main.py | 128 ++++++++++++++++++++++++++++++++++++++++--- ta-engine/main.py | 37 +++++++++---- 3 files changed, 215 insertions(+), 18 deletions(-) create mode 100644 docs/ai_org.md diff --git a/docs/ai_org.md b/docs/ai_org.md new file mode 100644 index 0000000..8ed70d1 --- /dev/null +++ b/docs/ai_org.md @@ -0,0 +1,68 @@ +# AI 주식 투자 회사 — 마스터 기획 (데이터 우선 재설계) + +> 사용자 의도: AI 에이전트들이 매일 회사처럼 스스로 운영. 사용자는 결과+승인. +> **핵심 원칙(재정립)**: 지능·자동실행보다 **데이터 신뢰성과 정확도 검증이 먼저**. +> "쓰레기 입력 → 쓰레기 추천". 토대가 검증되기 전엔 위층을 신뢰하지 않는다. +> 선택: 두뇌=하이브리드(EXAONE 1차 + Gemini 고급검증) / 승인=자동실행+사후보고 / 풀 조직. + +## 왜 순서를 바꿨나 (2026-06-03 발견된 토대 결함) +- 시세: 키움 토큰 race로 stale → 복구함. +- 일봉(stock_ohlcv): 06-02 366종목/오늘 0개. ka10005(30봉 한계)로 백필 안 됨 → **네이버 siseJson 경로 필요**. +- TA: `/analyze/all`이 dart_corps LIMIT 500 중 163개만, `is_active=true` 필터 없음. +- 정확도: 매일 "당일 실측 대비" 검증이 운영 중심이 아님. +→ 이 상태로 에이전트 조직을 올리면 추천이 오염된다. 데이터·검증을 먼저 세운다. + +--- + +## 로드맵 (각 Phase는 "검증 통과"가 다음 진행 조건) + +### Phase 0 — 데이터 신뢰 기반 (지금 시작) +목표: 모든 핵심 데이터가 **매일 최신·전종목·무결**하고, 깨지면 **즉시 감지**. +- [ ] OHLCV 백필 정상화: 네이버 siseJson 경로로 06-02~당일 + KOSPI 지수 채움. 키움은 보조. +- [ ] TA 커버리지: 전 활성종목 대상 + `is_active=true` 필터. +- [ ] **데이터 무결성 모니터**(신규): 시세/일봉/TA/점수/뉴스의 신선도·종목수·결측을 매시간 점검, + 기준 미달이면 텔레그램 경고 1건. (= "데이터 제대로 쌓이는지" 상시 감시) +- 검증: 모니터가 "전부 GREEN" 며칠 유지. LS 등 표본 종목 시세=일봉=TA=점수 일치. + +### Phase 1 — 정확도 검증 하베스트 (백본) +목표: 모든 예측을 **매일 당일 실측과 대조**해 맞는지 수치로 안다. +- [ ] 일일 검증 잡: 어제 추천/예측 vs 오늘 실제 등락 → 등급별 적중률·평균수익·알파·IC(스피어만) 기록. + (기존 verify_predictions_job / /backtest / /learn-weights 를 이 백본으로 통합·정규화) +- [ ] **핫종목 검증팀**을 여기 편입: 키움 핫종목(ka10023 거래량급증 등) × 시스템 추천 교차 → + 노이즈(작전·테마·일시급등) 필터 → 며칠 뒤 실측으로 "노이즈 판별이 맞았나" 정확도 추적. +- [ ] 정확도 리포트(텔레그램/대시보드): 방식이 맞는지 매일 사용자에게 숫자로 보고. +- 검증: 적중률·IC가 무작위(0) 대비 유의미하게 양(+)인지. 아니면 방식 수정. + +### Phase 2 — AI 조직 (신뢰 데이터 위에서) +- [ ] 🔍 리서치 2명: `run_deep_analysis(model="hybrid")` 재사용 (EXAONE↔Gemini 교차검증). +- [ ] ⚠️ 리스크 매니저: 사이징·섹터집중·레짐·분식·단기약세·일일손실 게이트 정식화. +- [ ] 🧭 CIO(신규, Gemini): 점수+논거+리스크+핫검증 종합 → "오늘의 결정안"(매수/매도/보류+확신도+사이즈+논거). +- 비용 가드: EXAONE 전종목 1차 → 상위 N(기본10)만 Gemini. `gemini_call_log` 일일한도 재사용. +- 검증: 결정안을 1~2주 dry-run, Phase 1 하베스트로 실측 적중 확인. + +### Phase 3 — 자동실행 + 사후보고 +- 기본 **모의투자 유지**(`KIS_IS_PAPER=true`). 실계좌 전환은 별도 명시 승인. +- 사전승인 필수("중요") 기본: ①1회 매수 ≥50만원 ②신규종목 ③일일손실 -2% 근접. +- 그 외 가드 내 자동실행 + 텔레그램 사후보고. 기존 7가드·halt 유지. +- 검증: 모의계좌 체결·손익이 결정안과 일치, 가드가 실제로 막는지. + +--- + +## 조직도 (Phase 2에서 완성, 기존 재사용 위주) +| 직책 | 구현 | 상태 | +|------|------|------| +| 📥 데이터팀 | 서비스 7종 | ✅ (Phase 0에서 신뢰성 보강) | +| 🔍 리서치 ×2 | run_deep_analysis(hybrid) | ✅ 재사용 | +| 💰 퀀트 | score-engine 10공식 | ✅ | +| 🔥 핫종목 검증팀 | 키움 핫 × 추천 노이즈필터 | 🟡 Phase 1 | +| ⚠️ 리스크 매니저 | 가드 묶음 | 🟡 Phase 2 | +| 🧭 CIO | 종합 결정안 (Gemini) | ❌ Phase 2 | +| 📋 리포터 / ✅ 사용자 | 텔레그램 보고·승인 | 🟡 Phase 3 | + +## 신규 산출물(누적) +- `data_health` 점검 모듈 + 신선도 모니터 (Phase 0) +- OHLCV siseJson 백필 보강 (Phase 0) +- `prediction_accuracy` 검증 하베스트 + 리포트 (Phase 1) +- `/hot/validate` 핫종목 노이즈필터 (Phase 1) +- `daily_decisions` 테이블 + CIO 로직 (Phase 2) +- 자동실행/사후보고 배선 (Phase 3) diff --git a/score-engine/main.py b/score-engine/main.py index 8268ec3..00a08ae 100644 --- a/score-engine/main.py +++ b/score-engine/main.py @@ -2061,8 +2061,10 @@ def calc_valuation_percentile(per_history: list, cur_per: float) -> tuple[float, return 0.0, "" -async def calculate_daily_scores(as_of: date | None = None): +async def calculate_daily_scores(as_of: date | None = None, notify: bool = False): """일간 점수 계산. as_of=None이면 today (운영 모드), as_of=date이면 그 시점 기준 (백필 모드). + notify=True일 때만 텔레그램 일간리포트/신규강력매수 발신 (16:30 정기 1회만 True). + 그 외 호출(통합 워크플로우 30분마다·매시간 스코어·18:30 등)은 silent로 점수만 갱신. 백필 모드는 look-ahead bias 차단: - 사후 학습 캐시(reliability/source_credibility) 미적용 - weight_config는 config_date <= as_of 필터 @@ -2818,7 +2820,7 @@ async def calculate_daily_scores(as_of: date | None = None): logger.info("scoring.done", scored=scored, recommended=len(rec_rows)) # 신규 강력매수 즉시 알림 (어제는 강력매수 아니었는데 오늘 새로 등장한 종목) - if not backfill_mode and strong_buy: + if notify and not backfill_mode and strong_buy: try: async with pg_pool.acquire() as nc: prev_codes = {r["stock_code"] for r in await nc.fetch(""" @@ -2853,8 +2855,8 @@ async def calculate_daily_scores(as_of: date | None = None): except Exception as e: logger.warning("new_buy.err", error=str(e)) - # 텔레그램 알림 - if strong_buy or strong_sell: + # 텔레그램 알림 (notify=True 정기 1회만 — 매 호출 발신 방지) + if notify and (strong_buy or strong_sell): lines = [f"📊 Trading AI 일간 리포트 ({today})\n"] if strong_buy: lines.append("🟢 강력매수 추천 (버핏 가치필터 통과)") @@ -3027,7 +3029,7 @@ async def startup(): redis_cl = aioredis.Redis( host=REDIS_HOST, port=6379, password=REDIS_PASSWORD, db=3, decode_responses=True) await init_db() - scheduler.add_job(calculate_daily_scores, "cron", + scheduler.add_job(lambda: calculate_daily_scores(notify=True), "cron", day_of_week="mon-fri", hour=16, minute=30, id="daily_score", replace_existing=True) # 텔레그램 정기 알림 하루 2회로 축소 (사용자 요청 2026-06-02): @@ -3038,6 +3040,14 @@ async def startup(): scheduler.add_job(cleanup_old_data, "cron", hour=4, minute=0, id="cleanup", replace_existing=True) + # 데이터 무결성 모니터: 평일 10~17시 매시간 (RED면 텔레그램 경고, 3h 쓰로틀) + scheduler.add_job(data_health_monitor_job, "cron", + day_of_week="mon-fri", hour="10-17", minute=5, + id="data_health", replace_existing=True) + # 정확도 검증 리포트: 매주 일요일 10시 (방식이 실측 대비 맞는지) + scheduler.add_job(accuracy_report_job, "cron", + day_of_week="sun", hour=10, minute=0, + id="accuracy_report", replace_existing=True) # 성과 추적: 매일 18시 가격 업데이트 scheduler.add_job(update_performance_prices, "cron", day_of_week="mon-fri", hour=18, minute=0, @@ -3144,13 +3154,117 @@ async def cleanup_old_data(): prices=deleted_prices, recs=deleted_recs, news=deleted_news, signals=deleted_signals) +# ── 데이터 무결성 모니터 ("데이터 제대로 쌓이는지" 상시 감시) ────────── +_last_health_alert: dict = {} # 동일 결함 반복경고 방지 (3시간 쓰로틀) + +async def check_data_health() -> dict: + """핵심 데이터 신선도·커버리지 점검 → 항목별 GREEN/YELLOW/RED.""" + checks = [] + def add(name, status, detail): checks.append({"name": name, "status": status, "detail": detail}) + async with pg_pool.acquire() as c: + n = await c.fetchval("SELECT COUNT(DISTINCT stock_code) FROM stock_prices WHERE collected_at::date=CURRENT_DATE") or 0 + add("시세(stock_prices)", "GREEN" if n>=2000 else "YELLOW" if n>=500 else "RED", f"오늘 {n}종목") + + last_dt = await c.fetchval("SELECT MAX(dt) FROM stock_ohlcv WHERE stock_code<>'KOSPI'") + ocnt = await c.fetchval("SELECT COUNT(DISTINCT stock_code) FROM stock_ohlcv WHERE dt=$1", last_dt) if last_dt else 0 + oage = (date.today()-last_dt).days if last_dt else 999 + add("일봉(stock_ohlcv)", "GREEN" if (ocnt>=2000 and oage<=4) else "YELLOW" if ocnt>=1000 else "RED", f"{last_dt} {ocnt}종목") + + n = await c.fetchval("SELECT COUNT(*) FROM stock_technical WHERE analyzed_at::date=CURRENT_DATE") or 0 + add("기술분석(TA)", "GREEN" if n>=1500 else "YELLOW" if n>=300 else "RED", f"오늘 {n}건") + + n = await c.fetchval("SELECT COUNT(*) FROM stock_scores WHERE score_date=CURRENT_DATE") or 0 + add("점수(stock_scores)", "GREEN" if n>=1000 else "YELLOW" if n>=100 else "RED", f"오늘 {n}종목") + + n = await c.fetchval("SELECT COUNT(*) FROM news_analysis WHERE analyzed_at > now()-interval '24 hours'") or 0 + add("뉴스(24h)", "GREEN" if n>=100 else "YELLOW" if n>=10 else "RED", f"{n}건") + + kd = await c.fetchval("SELECT MAX(dt) FROM stock_ohlcv WHERE stock_code='KOSPI'") + kage = (date.today()-kd).days if kd else 999 + add("KOSPI지수 일봉", "GREEN" if kage<=4 else "RED", f"{kd}") + worst = "RED" if any(x["status"]=="RED" for x in checks) else ("YELLOW" if any(x["status"]=="YELLOW" for x in checks) else "GREEN") + return {"overall": worst, "checks": checks, "checked_at": datetime.now().isoformat()} + +async def data_health_monitor_job(): + """평일 장중~마감후 점검. RED 있으면 텔레그램 경고 1건 (3시간 쓰로틀).""" + try: + h = await check_data_health() + except Exception as e: + logger.error("data_health.err", error=str(e)); return + reds = [x for x in h["checks"] if x["status"]=="RED"] + if not reds: return + key = ",".join(sorted(x["name"] for x in reds)) + now = datetime.now() + last = _last_health_alert.get(key) + if last and (now-last).total_seconds() < 10800: return + _last_health_alert[key] = now + lines = ["🚨 데이터 무결성 경고"] + [f"❌ {x['name']}: {x['detail']}" for x in reds] + lines.append("\n수집/분석 파이프라인 점검 필요") + await send_telegram("\n".join(lines)) + logger.info("data_health.alert", reds=len(reds)) + +@app.get("/data-health") +async def data_health_endpoint(): + return await check_data_health() + +# ── 정확도 검증 하베스트 ("방식이 맞는지" 실측 대비) ────────────────── +async def compute_accuracy(days: int = 90) -> dict: + """추천 등급별 사후 정확도. recommendation_performance(실측 7d/30d 수익률·알파) 집계. + 매수계열 알파>0 & 매도계열 알파<0 이면 방식 유효.""" + async with pg_pool.acquire() as conn: + grades = await conn.fetch(""" + SELECT recommendation rec, COUNT(*) n, + AVG(return_7d) ret7, AVG(alpha_7d) a7, + AVG(return_30d) ret30, AVG(alpha_30d) a30, + AVG(CASE WHEN return_7d>0 THEN 1.0 ELSE 0 END) up7 + FROM recommendation_performance + WHERE return_7d IS NOT NULL AND rec_date >= CURRENT_DATE - ($1::int) + GROUP BY recommendation + """, days) + order = {"강력매수": 0, "매수관심": 1, "관망": 2, "매도관심": 3, "강력매도": 4} + rows = sorted([dict(g) for g in grades], key=lambda x: order.get(x["rec"], 9)) + def wavg(sel, key): + tot = sum(r["n"] for r in rows if r["rec"] in sel) + s = sum((r[key] or 0) * r["n"] for r in rows if r["rec"] in sel) + return round(s / tot, 2) if tot else None + buy_a, sell_a = wavg(("강력매수", "매수관심"), "a7"), wavg(("강력매도", "매도관심"), "a7") + ok = (buy_a is not None and sell_a is not None and buy_a > 0 and sell_a < 0) + return { + "days": days, + "grades": [{"rec": r["rec"], "n": r["n"], + "ret7": round(r["ret7"] or 0, 2), "alpha7": round(r["a7"] or 0, 2), + "ret30": round(r["ret30"], 2) if r["ret30"] is not None else None, + "alpha30": round(r["a30"], 2) if r["a30"] is not None else None, + "up7_pct": round(100 * (r["up7"] or 0))} for r in rows], + "buy_alpha7": buy_a, "sell_alpha7": sell_a, + "verdict": "양호 (매수 알파>0, 매도 알파<0)" if ok else "교정필요 (매수/매도 변별력 부족)", + } + +@app.get("/accuracy") +async def accuracy_endpoint(days: int = Query(default=90, ge=7, le=365)): + return await compute_accuracy(days) + +async def accuracy_report_job(): + """주간 정확도 리포트 — 방식이 실측 대비 맞는지 텔레그램 보고.""" + try: + a = await compute_accuracy(90) + except Exception as e: + logger.error("accuracy.err", error=str(e)); return + lines = ["📈 추천 정확도 리포트 (최근90일·7일 알파)", + f"판정: {a['verdict']}", + f"매수계열 알파 {a['buy_alpha7']} / 매도계열 알파 {a['sell_alpha7']}\n"] + for g in a["grades"]: + lines.append(f"{g['rec']}: n{g['n']} 수익{g['ret7']}% 알파{g['alpha7']}% 상승{g['up7_pct']}%") + await send_telegram("\n".join(lines)) + logger.info("accuracy.report.sent", verdict=a["verdict"]) + @app.get("/health") async def health(): return {"status": "ok"} @app.post("/score/calculate") -async def manual_calc(): - n = await calculate_daily_scores() +async def manual_calc(notify: bool = Query(default=False)): + n = await calculate_daily_scores(notify=notify) return {"status": "done", "scored": n} diff --git a/ta-engine/main.py b/ta-engine/main.py index 6b01a92..e91ac90 100644 --- a/ta-engine/main.py +++ b/ta-engine/main.py @@ -713,16 +713,31 @@ async def init_db(): # ── 전체 분석 작업 ──────────────────────────────────────── -async def job_analyze(): - logger.info("ta.job.start") +async def job_analyze(limit: int = 500): + """limit>0: 시총 상위 N개(장중 경량). limit=0: 전 활성종목(장마감 풀커버). + is_active=true 필터 필수 — 상장폐지 제외 + LS 등 누락 방지.""" + logger.info("ta.job.start", limit=limit) async with httpx.AsyncClient() as client: codes: List[tuple] = [] if pg_pool: try: - rows = await pg_pool.fetch("SELECT stock_code, corp_name FROM dart_corps LIMIT 500") + q = """ + SELECT c.stock_code, c.corp_name + FROM dart_corps c + LEFT JOIN ( + SELECT DISTINCT ON (stock_code) stock_code, market_cap + FROM stock_prices ORDER BY stock_code, collected_at DESC + ) p ON p.stock_code = c.stock_code + WHERE c.is_active = true + ORDER BY COALESCE(p.market_cap, 0) DESC + """ + if limit and limit > 0: + q += f" LIMIT {int(limit)}" + rows = await pg_pool.fetch(q) codes = [(r["stock_code"], r["corp_name"] or "") for r in rows if r["stock_code"]] - except: pass + except Exception as e: + logger.warning("ta.codes.err", error=str(e)) if not codes: for sosok in [0, 1]: @@ -740,7 +755,7 @@ async def job_analyze(): if len(codes) >= 500: break ok = 0 - for code, name in codes[:500]: + for code, name in codes: if not code or len(code) != 6: continue try: result = await analyze_stock(client, code, name) @@ -748,11 +763,11 @@ async def job_analyze(): except Exception as e: stats.errors += 1 logger.warning("ta.analyze.err", code=code, error=str(e)) - await asyncio.sleep(0.4) + await asyncio.sleep(0.2) stats.analyzed += ok stats.last_run = datetime.now().isoformat() - logger.info("ta.job.done", analyzed=ok) + logger.info("ta.job.done", analyzed=ok, requested=len(codes)) # ── FastAPI ──────────────────────────────────────────────── @@ -768,9 +783,9 @@ async def startup(): redis_cl = aioredis.Redis( host=REDIS_HOST, port=6379, password=REDIS_PASSWORD, db=5, decode_responses=True) await init_db() - scheduler.add_job(job_analyze, "cron", day_of_week="mon-fri", + scheduler.add_job(lambda: job_analyze(limit=500), "cron", day_of_week="mon-fri", hour="9-16", minute="*/30", id="ta_30m", replace_existing=True) - scheduler.add_job(job_analyze, "cron", day_of_week="mon-fri", + scheduler.add_job(lambda: job_analyze(limit=0), "cron", day_of_week="mon-fri", hour=16, minute=15, id="ta_close", replace_existing=True) scheduler.start() logger.info("ta-engine.started") @@ -856,8 +871,8 @@ async def buy_candidates(limit: int = Query(default=20)): return result @app.post("/analyze/all") -async def analyze_all(): - asyncio.create_task(job_analyze()) +async def analyze_all(limit: int = 0): + asyncio.create_task(job_analyze(limit=limit)) return {"status": "started"} @app.post("/analyze/{code}")