US2KR/us2kr/pipeline.py

"""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
from us2kr.render import render_markdown
from us2kr.store import Store
from us2kr.summarize import build_provider, summarize_payload
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 _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
        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

        markdown = render_markdown(payload, report, model_used)
        result = BriefingResult(
            payload=payload,
            report=report,
            markdown=markdown,
            model_used=model_used,
            degraded=degraded,
            error=error,
        )
        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()