feat: 데이터 토대 + 정확도 검증 (AI 투자조직 Phase 0~1)

- 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) <noreply@anthropic.com>
This commit is contained in:
kyu
2026-06-03 12:29:02 +09:00
parent a56227ace9
commit 5c721e6b70
3 changed files with 215 additions and 18 deletions
+26 -11
View File
@@ -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}")