jsahnandClaude Fable 5
ed48f4c96b
After summarisation the report is verified against the collected payload (numbers, directions, unsupported claims) by a second LLM call with a Verification JSON schema. Major issues trigger one revision and a re-check; minor-only verdicts are normalized to ok. Failures never block delivery. The verdict is stored in the result JSON and shown on the dashboard (badge + 검증 메모). Disable with US2KR_VERIFY_ENABLED. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
318 lines
13 KiB
Python
318 lines
13 KiB
Python
"""Briefing pipeline: collect -> store -> summarise -> render -> deliver.
|
|
|
|
All collaborators are imported at module level and referenced as module attributes so that tests
|
|
can monkeypatch e.g. ``us2kr.pipeline.fetch_quotes``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from collections.abc import Sequence
|
|
from datetime import date, datetime, timedelta
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from us2kr.collectors.base import run_safely
|
|
from us2kr.collectors.events import fetch_earnings, fetch_events
|
|
from us2kr.collectors.news import dedup_news, fetch_news
|
|
from us2kr.collectors.prices import fetch_quotes
|
|
from us2kr.collectors.searx import fetch_searx_news
|
|
from us2kr.config import KST, Settings
|
|
from us2kr.deliver import build_deliverers
|
|
from us2kr.llm.base import LLMError
|
|
from us2kr.market_time import MarketClock
|
|
from us2kr.models import BriefingResult, MarketPayload, Report, Verification
|
|
from us2kr.render import render_markdown
|
|
from us2kr.store import Store
|
|
from us2kr.summarize import build_provider, summarize_payload, verify_and_revise
|
|
from us2kr.symbols import SYMBOL_GROUPS
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
EARNINGS_GROUPS: tuple[str, ...] = ("semis", "megacap", "korea_linked")
|
|
# ETFs never report earnings; querying them only burns the earnings time budget.
|
|
EARNINGS_EXCLUDE: frozenset[str] = frozenset({"EWY"})
|
|
SKIP_ALREADY_SENT = "already sent"
|
|
SKIP_KRX_CLOSED = "krx closed"
|
|
DEGRADED_SUBJECT = "LLM 요약 실패 - 축소 모드 발송"
|
|
DELIVERY_FAILED_SUBJECT = "브리핑 발송 실패"
|
|
DELIVERY_FAILED_PREFIX = "delivery failed"
|
|
|
|
|
|
def _now() -> datetime:
|
|
return datetime.now(ZoneInfo(KST))
|
|
|
|
|
|
def earnings_tickers() -> list[str]:
|
|
"""Tickers whose earnings we track (semis + megacap + Korea-linked), de-duplicated."""
|
|
seen: dict[str, None] = {}
|
|
for group in EARNINGS_GROUPS:
|
|
for ticker in SYMBOL_GROUPS.get(group, {}):
|
|
if ticker not in EARNINGS_EXCLUDE:
|
|
seen.setdefault(ticker, None)
|
|
return list(seen)
|
|
|
|
|
|
def _to_json(items: Sequence[BaseModel]) -> str:
|
|
return json.dumps([m.model_dump(mode="json") for m in items], ensure_ascii=False, default=str)
|
|
|
|
|
|
def _save_raw(store: Store, session: date, name: str, items: Sequence[BaseModel]) -> None:
|
|
"""Archive a collector's raw output; failures are logged but never fatal."""
|
|
try:
|
|
store.save_raw(session, name, _to_json(items))
|
|
except Exception: # archiving is best-effort
|
|
log.exception("save_raw(%s) failed", name)
|
|
|
|
|
|
def build_payload(session: date, settings: Settings, clock: MarketClock, store: Store) -> MarketPayload:
|
|
"""Run every collector with failure isolation and assemble the MarketPayload."""
|
|
now = _now()
|
|
us_close_time = clock.us_session_close(session)
|
|
krx_date = clock.krx_date_for(session)
|
|
krx_open_today = clock.is_krx_session(krx_date)
|
|
gaps: list[str] = []
|
|
|
|
quotes = run_safely("prices", lambda: fetch_quotes(session), [], gaps)
|
|
news_since = us_close_time - timedelta(hours=settings.news_lookback_hours)
|
|
news = run_safely(
|
|
"news", lambda: fetch_news(since=news_since, max_items=settings.news_max_items), [], gaps
|
|
)
|
|
if settings.searxng_base_url:
|
|
searx_news = run_safely(
|
|
"searxng",
|
|
lambda: fetch_searx_news(
|
|
settings.searxng_base_url, # type: ignore[arg-type]
|
|
since=news_since,
|
|
queries=settings.searxng_queries or None,
|
|
),
|
|
[],
|
|
gaps,
|
|
)
|
|
if searx_news:
|
|
# RSS first so it wins timestamp ties in dedup; cap stays at news_max_items overall.
|
|
news = dedup_news([*news, *searx_news])[: settings.news_max_items]
|
|
events = run_safely("events", lambda: fetch_events(session, settings.fred_api_key), [], gaps)
|
|
tickers = earnings_tickers()
|
|
earnings = run_safely("earnings", lambda: fetch_earnings(session, tickers), [], gaps)
|
|
|
|
if quotes:
|
|
run_safely("store", lambda: store.save_quotes(session, quotes), None, gaps)
|
|
elif not any(g.startswith("prices:") for g in gaps):
|
|
gaps.append("prices: no quotes")
|
|
history_notes = run_safely("history", lambda: store.history_notes(session, quotes), [], gaps)
|
|
|
|
_save_raw(store, session, "prices", quotes)
|
|
_save_raw(store, session, "news", news)
|
|
_save_raw(store, session, "events", events)
|
|
_save_raw(store, session, "earnings", earnings)
|
|
|
|
payload = MarketPayload(
|
|
session_date=session,
|
|
us_close_time=us_close_time,
|
|
generated_at=now,
|
|
krx_date=krx_date,
|
|
krx_open_today=krx_open_today,
|
|
quotes=quotes,
|
|
news=news,
|
|
events=events,
|
|
earnings=earnings,
|
|
history_notes=history_notes,
|
|
data_gaps=gaps,
|
|
)
|
|
log.info(
|
|
"payload built for %s: %d quotes, %d news, %d events, %d earnings, %d gaps",
|
|
session,
|
|
len(quotes),
|
|
len(news),
|
|
len(events),
|
|
len(earnings),
|
|
len(gaps),
|
|
)
|
|
return payload
|
|
|
|
|
|
def _minimal_payload(session: date, clock: MarketClock, krx_date: date, krx_open: bool, now: datetime):
|
|
return MarketPayload(
|
|
session_date=session,
|
|
us_close_time=clock.us_session_close(session),
|
|
generated_at=now,
|
|
krx_date=krx_date,
|
|
krx_open_today=krx_open,
|
|
)
|
|
|
|
|
|
def _skip_result(payload: MarketPayload, reason: str) -> BriefingResult:
|
|
return BriefingResult(payload=payload, report=None, markdown="", error=reason)
|
|
|
|
|
|
def _summarize(payload: MarketPayload, settings: Settings) -> tuple[Report | None, str | None, str | None]:
|
|
"""Return (report, model_used, error). Any failure degrades instead of aborting the briefing."""
|
|
try:
|
|
provider = build_provider(settings)
|
|
report, model_used = summarize_payload(payload, provider)
|
|
return report, model_used, None
|
|
except LLMError as exc:
|
|
log.error("LLM summarisation failed, switching to degraded mode: %s", exc)
|
|
return None, None, str(exc)
|
|
except Exception as exc: # provider construction / unexpected SDK errors
|
|
log.exception("unexpected LLM failure, switching to degraded mode")
|
|
return None, None, f"{type(exc).__name__}: {exc}"
|
|
|
|
|
|
def _verify(
|
|
payload: MarketPayload, report: Report, settings: Settings
|
|
) -> tuple[Report, Verification | None]:
|
|
"""Fact-check (and possibly revise) the report. Any failure keeps the report and returns None."""
|
|
try:
|
|
provider = build_provider(settings)
|
|
return verify_and_revise(payload, report, provider)
|
|
except Exception as exc: # noqa: BLE001 - verification must never block the briefing
|
|
log.warning("verification failed, delivering unverified report: %s: %s", type(exc).__name__, exc)
|
|
return report, None
|
|
|
|
|
|
def _superseded_before_krx_open(session: date, krx_date: date, clock: MarketClock) -> bool:
|
|
"""True when another US session closes before KRX next trades, so this briefing would be stale.
|
|
|
|
Friday session -> Saturday krx_date: next KRX = Monday, next US session = Monday -> NOT superseded
|
|
(Monday's open needs Friday's recap). Chuseok Wed 09-23 -> Thu 09-24 closed: next KRX Mon 09-28,
|
|
next US Thu 09-24 -> superseded -> skip.
|
|
"""
|
|
return clock.next_us_session(session) < clock.next_krx_session(krx_date)
|
|
|
|
|
|
def _notify_all(deliverers: Sequence[object], subject: str, detail: str) -> None:
|
|
for deliverer in deliverers:
|
|
try:
|
|
deliverer.notify_error(subject, detail) # type: ignore[attr-defined]
|
|
except Exception: # notify_error must never propagate
|
|
log.exception("notify_error via %s failed", getattr(deliverer, "name", deliverer))
|
|
|
|
|
|
def _deliver(result: BriefingResult, settings: Settings, store: Store, report_id: int) -> list[str]:
|
|
"""Deliver on every channel. Returns the names of channels that failed (empty on success).
|
|
|
|
Failures are isolated per channel but never silent: ``result.error`` is annotated (see
|
|
``delivery_failed``) and an operator alert goes through every channel - FileDeliverer always writes
|
|
errors.log, and a broken Telegram channel may still accept the short plain-text alert.
|
|
"""
|
|
deliverers = build_deliverers(settings)
|
|
failures: list[tuple[str, str]] = []
|
|
for deliverer in deliverers:
|
|
name = getattr(deliverer, "name", str(deliverer))
|
|
try:
|
|
receipt = deliverer.deliver(result)
|
|
store.mark_sent(report_id, name, receipt)
|
|
log.info("delivered via %s: %s", name, receipt)
|
|
except Exception as exc: # one channel failing must not block the others
|
|
log.exception("delivery via %s failed", name)
|
|
failures.append((name, f"{type(exc).__name__}: {exc}"[:300]))
|
|
|
|
if result.degraded and result.error:
|
|
_notify_all(deliverers, DEGRADED_SUBJECT, result.error)
|
|
|
|
if failures:
|
|
detail = "; ".join(f"{name}: {err}" for name, err in failures)
|
|
note = f"{DELIVERY_FAILED_PREFIX}: {detail}"
|
|
result.error = f"{result.error} | {note}" if result.error else note
|
|
log.error("%s for KRX %s (report id=%d)", note, result.payload.krx_date, report_id)
|
|
_notify_all(
|
|
deliverers,
|
|
DELIVERY_FAILED_SUBJECT,
|
|
f"KRX {result.payload.krx_date} (report id={report_id}) - {detail} - "
|
|
f"재발송: us2kr run-once --force --date {result.payload.session_date.isoformat()}",
|
|
)
|
|
return [name for name, _ in failures]
|
|
|
|
|
|
def delivery_failed(result: BriefingResult) -> bool:
|
|
"""True when at least one delivery channel failed for this result."""
|
|
return bool(result.error) and DELIVERY_FAILED_PREFIX in (result.error or "")
|
|
|
|
|
|
def run_briefing(
|
|
session: date | None,
|
|
settings: Settings,
|
|
*,
|
|
dry_run: bool = False,
|
|
force: bool = False,
|
|
no_llm: bool = False,
|
|
) -> BriefingResult:
|
|
"""Produce and deliver one briefing for ``session`` (default: latest completed US session)."""
|
|
clock = MarketClock(settings.run_offset_minutes, extra_krx_holidays=settings.krx_extra_holidays)
|
|
store = Store(settings.db_path)
|
|
try:
|
|
now = _now()
|
|
if session is None:
|
|
session = clock.latest_completed_session(now)
|
|
elif not clock.is_us_session(session):
|
|
raise ValueError(f"{session} is not a US trading session")
|
|
|
|
krx_date = clock.krx_date_for(session)
|
|
krx_open = clock.is_krx_session(krx_date)
|
|
log.info("briefing for US session %s -> KRX date %s (krx open: %s)", session, krx_date, krx_open)
|
|
close_at = clock.us_session_close(session)
|
|
if close_at > now:
|
|
# A not-yet-closed session must never reach save_quotes/save_report: the stored report would make
|
|
# the daemon's scheduled post-close run (and catch-up) skip as "already sent". --force does not
|
|
# bypass this (it means "resend an already-sent briefing"); test delivery with a past session.
|
|
if not dry_run:
|
|
close_txt = close_at.isoformat(timespec="minutes")
|
|
raise ValueError(
|
|
f"US session {session} has not closed yet (close {close_txt}); refusing to save/deliver "
|
|
"a partial briefing. Use --dry-run for an intraday preview, or --date <previous session> "
|
|
"--force to test delivery."
|
|
)
|
|
log.warning("US session %s has not closed yet - dry-run preview with intraday quotes", session)
|
|
|
|
if settings.skip_when_krx_closed and not krx_open:
|
|
if _superseded_before_krx_open(session, krx_date, clock):
|
|
log.info("KRX closed on %s; a later US session closes before it reopens - skipping", krx_date)
|
|
payload = _minimal_payload(session, clock, krx_date, krx_open, now)
|
|
return _skip_result(payload, SKIP_KRX_CLOSED)
|
|
log.info("KRX closed on %s but this is the last US session before it reopens - sending", krx_date)
|
|
if not dry_run and not force and store.has_report(krx_date):
|
|
log.info("briefing for %s already sent - skipping (use --force to resend)", krx_date)
|
|
return _skip_result(_minimal_payload(session, clock, krx_date, krx_open, now), SKIP_ALREADY_SENT)
|
|
|
|
payload = build_payload(session, settings, clock, store)
|
|
|
|
report: Report | None = None
|
|
model_used: str | None = None
|
|
error: str | None = None
|
|
verification: Verification | None = None
|
|
degraded = False
|
|
if no_llm:
|
|
degraded = True
|
|
log.info("--no-llm: skipping summarisation")
|
|
else:
|
|
report, model_used, error = _summarize(payload, settings)
|
|
degraded = report is None
|
|
if report is not None and settings.verify_enabled:
|
|
report, verification = _verify(payload, report, settings)
|
|
|
|
markdown = render_markdown(payload, report, model_used)
|
|
result = BriefingResult(
|
|
payload=payload,
|
|
report=report,
|
|
markdown=markdown,
|
|
model_used=model_used,
|
|
degraded=degraded,
|
|
error=error,
|
|
verification=verification,
|
|
)
|
|
if dry_run:
|
|
print(markdown)
|
|
return result
|
|
|
|
# Save before delivering: a crash mid-delivery must not cause a duplicate on catch-up.
|
|
report_id = store.save_report(result)
|
|
log.info("report %d saved for %s (degraded=%s)", report_id, krx_date, degraded)
|
|
_deliver(result, settings, store, report_id)
|
|
return result
|
|
finally:
|
|
store.close()
|