215 lines
6.3 KiB
Python
215 lines
6.3 KiB
Python
|
import json
|
||
|
|
from datetime import date
|
||
|
|
|
||
|
|
from fastapi import APIRouter, HTTPException
|
||
|
|
|
||
|
|
from ..services.llm_extract import extract_event
|
||
|
|
from ..services.market_moves import fetch_moves_via_qfr
|
||
|
|
from ..services.store import (
|
||
|
|
ensure_schema,
|
||
|
|
get_store,
|
||
|
|
insert_event_result,
|
||
|
|
insert_raw_item,
|
||
|
|
)
|
||
|
|
|
||
|
|
router = APIRouter()
|
||
|
|
|
||
|
|
|
||
|
|
def _today() -> str:
|
||
|
|
return date.today().isoformat()
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/rss")
|
||
|
|
def ingest_rss(payload: dict):
|
||
|
|
"""Ingest one or many RSS items.
|
||
|
|
|
||
|
|
Expected payload:
|
||
|
|
{"items": [{"title":..., "url":..., "published_at":..., "summary":..., "lang":...}, ...]}
|
||
|
|
"""
|
||
|
|
items = payload.get("items")
|
||
|
|
if not isinstance(items, list) or not items:
|
||
|
|
raise HTTPException(status_code=400, detail="payload.items must be a non-empty list")
|
||
|
|
|
||
|
|
st = get_store()
|
||
|
|
conn = st.connect()
|
||
|
|
ensure_schema(conn)
|
||
|
|
|
||
|
|
n = 0
|
||
|
|
for it in items:
|
||
|
|
if not isinstance(it, dict):
|
||
|
|
continue
|
||
|
|
insert_raw_item(
|
||
|
|
conn,
|
||
|
|
source="rss",
|
||
|
|
item_date=payload.get("date") or _today(),
|
||
|
|
title=(it.get("title") or "")[:500],
|
||
|
|
content=(it.get("summary") or it.get("content") or "")[:20_000],
|
||
|
|
url=it.get("url"),
|
||
|
|
published_at=it.get("published_at"),
|
||
|
|
lang=it.get("lang"),
|
||
|
|
)
|
||
|
|
n += 1
|
||
|
|
|
||
|
|
return {"ok": True, "inserted": n}
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/macro")
|
||
|
|
def ingest_macro(payload: dict):
|
||
|
|
items = payload.get("items")
|
||
|
|
if not isinstance(items, list) or not items:
|
||
|
|
raise HTTPException(status_code=400, detail="payload.items must be a non-empty list")
|
||
|
|
|
||
|
|
st = get_store()
|
||
|
|
conn = st.connect()
|
||
|
|
ensure_schema(conn)
|
||
|
|
|
||
|
|
n = 0
|
||
|
|
for it in items:
|
||
|
|
if not isinstance(it, dict):
|
||
|
|
continue
|
||
|
|
insert_raw_item(
|
||
|
|
conn,
|
||
|
|
source="macro",
|
||
|
|
item_date=payload.get("date") or _today(),
|
||
|
|
title=(it.get("title") or "")[:500],
|
||
|
|
content=(it.get("content") or "")[:20_000],
|
||
|
|
url=it.get("url"),
|
||
|
|
published_at=it.get("published_at"),
|
||
|
|
lang=it.get("lang"),
|
||
|
|
)
|
||
|
|
n += 1
|
||
|
|
|
||
|
|
return {"ok": True, "inserted": n}
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/market_moves")
|
||
|
|
def ingest_market_moves(payload: dict):
|
||
|
|
items = payload.get("items")
|
||
|
|
if not isinstance(items, list) or not items:
|
||
|
|
raise HTTPException(status_code=400, detail="payload.items must be a non-empty list")
|
||
|
|
|
||
|
|
st = get_store()
|
||
|
|
conn = st.connect()
|
||
|
|
ensure_schema(conn)
|
||
|
|
|
||
|
|
n = 0
|
||
|
|
for it in items:
|
||
|
|
if not isinstance(it, dict):
|
||
|
|
continue
|
||
|
|
insert_raw_item(
|
||
|
|
conn,
|
||
|
|
source="market_moves",
|
||
|
|
item_date=payload.get("date") or _today(),
|
||
|
|
title=(it.get("title") or "")[:500],
|
||
|
|
content=(it.get("content") or "")[:20_000],
|
||
|
|
url=it.get("url"),
|
||
|
|
published_at=it.get("published_at"),
|
||
|
|
lang=it.get("lang"),
|
||
|
|
)
|
||
|
|
n += 1
|
||
|
|
|
||
|
|
return {"ok": True, "inserted": n}
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/market_moves/run")
|
||
|
|
def run_market_moves(payload: dict | None = None):
|
||
|
|
"""Generate daily market-move items from QFR raw data and parse them into events."""
|
||
|
|
|
||
|
|
payload = payload or {}
|
||
|
|
day = str(payload.get("date") or _today())
|
||
|
|
# QFR raw data uses trade_date like YYYYMMDD.
|
||
|
|
trade_date = day.replace("-", "")
|
||
|
|
|
||
|
|
st = get_store()
|
||
|
|
conn = st.connect()
|
||
|
|
ensure_schema(conn)
|
||
|
|
|
||
|
|
data = fetch_moves_via_qfr(trade_date=trade_date, symbols=payload.get("symbols"))
|
||
|
|
if not data.get("ok"):
|
||
|
|
raise HTTPException(status_code=500, detail=data)
|
||
|
|
|
||
|
|
inserted = 0
|
||
|
|
parsed_ok = 0
|
||
|
|
parsed_err = 0
|
||
|
|
|
||
|
|
for mv in data.get("moves", []):
|
||
|
|
sym = mv.get("symbol")
|
||
|
|
td = mv.get("trade_date")
|
||
|
|
ret_1d = mv.get("ret_1d")
|
||
|
|
vol_20d = mv.get("vol_20d")
|
||
|
|
z_1d = mv.get("z_1d")
|
||
|
|
|
||
|
|
title = f"Market move {sym} {td}: ret_1d={ret_1d:.4f} z_1d={z_1d:.2f}" if isinstance(ret_1d, (int, float)) and isinstance(z_1d, (int, float)) else f"Market move {sym} {td}"
|
||
|
|
content = (
|
||
|
|
f"symbol={sym}\n"
|
||
|
|
f"trade_date={td}\n"
|
||
|
|
f"prev_trade_date={mv.get('prev_trade_date')}\n"
|
||
|
|
f"close={mv.get('close')} prev_close={mv.get('prev_close')}\n"
|
||
|
|
f"ret_1d={ret_1d} vol_20d={vol_20d} z_1d={z_1d}\n"
|
||
|
|
"Interpretation task: explain the most likely macro/industry drivers for this move and which assets could be affected."
|
||
|
|
)
|
||
|
|
|
||
|
|
raw_item_id = insert_raw_item(
|
||
|
|
conn,
|
||
|
|
source="market_moves",
|
||
|
|
item_date=day,
|
||
|
|
title=title[:500],
|
||
|
|
content=content[:20_000],
|
||
|
|
url=None,
|
||
|
|
published_at=None,
|
||
|
|
lang="en",
|
||
|
|
)
|
||
|
|
inserted += 1
|
||
|
|
|
||
|
|
try:
|
||
|
|
res = extract_event(title=title, content=content, lang_hint="en")
|
||
|
|
except Exception as e:
|
||
|
|
# Network/provider errors should not abort the whole batch.
|
||
|
|
insert_event_result(
|
||
|
|
conn,
|
||
|
|
raw_item_id=raw_item_id,
|
||
|
|
model="",
|
||
|
|
ok=False,
|
||
|
|
event_json=None,
|
||
|
|
error=f"llm_exception:{type(e).__name__}",
|
||
|
|
)
|
||
|
|
parsed_err += 1
|
||
|
|
continue
|
||
|
|
|
||
|
|
if res.get("ok") is True:
|
||
|
|
insert_event_result(
|
||
|
|
conn,
|
||
|
|
raw_item_id=raw_item_id,
|
||
|
|
model=str(res.get("model") or ""),
|
||
|
|
ok=True,
|
||
|
|
event_json=json.dumps(res.get("event"), ensure_ascii=True),
|
||
|
|
error=None,
|
||
|
|
)
|
||
|
|
parsed_ok += 1
|
||
|
|
else:
|
||
|
|
err = str(res.get("error") or "unknown")
|
||
|
|
if err == "llm_failed":
|
||
|
|
# Keep a short hint to debug gateway flakiness without dumping secrets.
|
||
|
|
exc = str(res.get("exc") or "")
|
||
|
|
if exc:
|
||
|
|
err = f"{err}:{exc}"
|
||
|
|
insert_event_result(
|
||
|
|
conn,
|
||
|
|
raw_item_id=raw_item_id,
|
||
|
|
model=str(res.get("model") or ""),
|
||
|
|
ok=False,
|
||
|
|
event_json=None,
|
||
|
|
error=err,
|
||
|
|
)
|
||
|
|
parsed_err += 1
|
||
|
|
|
||
|
|
return {
|
||
|
|
"ok": True,
|
||
|
|
"date": day,
|
||
|
|
"inserted": inserted,
|
||
|
|
"parsed_ok": parsed_ok,
|
||
|
|
"parsed_err": parsed_err,
|
||
|
|
"errors": data.get("errors", []),
|
||
|
|
"symbols": data.get("symbols"),
|
||
|
|
}
|