"""Monday-morning weekly recap: aggregate the stored week into one briefing. No US session closes into Monday KST (the US Sunday never trades), so the Monday slot carries a recap of the previous week's US sessions instead. Everything is built from what the daily pipeline already stored - the `quotes` table for week-over-week changes and the daily reports for the narrative - so no collector runs. The result is a regular ``BriefingResult`` with ``kind="weekly"`` whose payload reuses ``MarketPayload``: ``quotes`` carry weekly changes (close = Friday close, prev_close = the previous week's close), ``history_notes`` carry the daily briefing headlines, ``session_date`` is the week's last US session and ``krx_date`` the Monday. That keeps the store, every deliverer and the dashboard working unchanged. """ from __future__ import annotations import json import logging from datetime import date, datetime, timedelta from zoneinfo import ZoneInfo from us2kr.config import KST, Settings from us2kr.llm.base import LLMError from us2kr.llm.prompts import WEEKLY_SYSTEM_PROMPT, build_weekly_user_message from us2kr.market_time import MarketClock from us2kr.models import BriefingResult, MarketPayload, Quote, Report, Verification from us2kr.pipeline import SKIP_ALREADY_SENT, _deliver, _skip_result from us2kr.render import render_weekly_markdown from us2kr.store import Store from us2kr.summarize import build_provider, summarize_weekly_payload, verify_and_revise from us2kr.symbols import display_name, group_of log = logging.getLogger(__name__) WEEKDAY_KO = ("월", "화", "수", "목", "금", "토", "일") def _now() -> datetime: return datetime.now(ZoneInfo(KST)) def most_recent_monday(today: date) -> date: """Today if it is a Monday, otherwise the latest past Monday.""" return today - timedelta(days=today.weekday()) def week_sessions_for(monday: date, clock: MarketClock) -> list[date]: """The previous week's US sessions (Mon..Fri before `monday`), holidays excluded.""" return [ d for d in (monday - timedelta(days=offset) for offset in range(7, 2, -1)) if clock.is_us_session(d) ] def _weekly_quote(symbol: str, rows: list) -> Quote: """Collapse one symbol's session rows (oldest first) into a week-over-week Quote.""" first, last = rows[0], rows[-1] base = first["prev_close"] if first["prev_close"] is not None else first["close"] end = last["close"] change_pct = None change_abs = None if end is not None and base: change_pct = (end / base - 1.0) * 100.0 change_abs = end - base highs = [r["day_high"] for r in rows if r["day_high"] is not None] lows = [r["day_low"] for r in rows if r["day_low"] is not None] return Quote( symbol=symbol, name=display_name(symbol), group=group_of(symbol) or "etc", close=end, prev_close=base, change_pct=change_pct, change_abs=change_abs, day_high=max(highs) if highs else None, day_low=min(lows) if lows else None, source="store", ) def _daily_headline_notes(store: Store, monday: date, gaps: list[str]) -> list[str]: """'미 {세션}({요일}): {헤드라인}' lines from the week's stored daily briefings.""" notes: list[str] = [] for row in store.reports_between(monday - timedelta(days=6), monday - timedelta(days=1)): try: data = json.loads(row["result_json"] or "{}") except json.JSONDecodeError: gaps.append(f"daily report {row['krx_date']}: unreadable") continue session = row["session_date"] or data.get("payload", {}).get("session_date") or "?" try: weekday = WEEKDAY_KO[date.fromisoformat(session).weekday()] label = f"미 {session}({weekday})" except ValueError: label = f"미 {session}" report = data.get("report") or {} headline = (report.get("headline") or "").strip() notes.append(f"{label}: {headline}" if headline else f"{label}: (요약 없음)") return notes def build_weekly_payload( monday: date, settings: Settings, clock: MarketClock, store: Store ) -> tuple[MarketPayload, list[date]]: """Assemble the weekly MarketPayload for the Monday `monday` from stored data only.""" sessions = week_sessions_for(monday, clock) gaps: list[str] = [] quotes: list[Quote] = [] covered: set[str] = set() if sessions: by_symbol: dict[str, list] = {} for row in store.load_quotes_between(sessions[0], sessions[-1]): by_symbol.setdefault(row["symbol"], []).append(row) covered.add(row["session_date"]) quotes = [_weekly_quote(symbol, rows) for symbol, rows in sorted(by_symbol.items())] missing = [s for s in sessions if s.isoformat() not in covered] if not quotes: gaps.append("weekly quotes: no stored sessions for last week") elif missing: gaps.append("weekly quotes: missing sessions " + ", ".join(s.isoformat() for s in missing)) history_notes = _daily_headline_notes(store, monday, gaps) if not history_notes: gaps.append("daily headlines: no stored briefings for last week") last_session = sessions[-1] if sessions else monday - timedelta(days=3) payload = MarketPayload( session_date=last_session, us_close_time=clock.us_session_close(last_session), generated_at=_now(), krx_date=monday, krx_open_today=clock.is_krx_session(monday), quotes=quotes, history_notes=history_notes, data_gaps=gaps, ) log.info( "weekly payload for %s: %d sessions, %d symbols, %d headlines, %d gaps", monday, len(sessions), len(quotes), len(history_notes), len(gaps), ) return payload, sessions def _summarize_weekly( payload: MarketPayload, sessions: list[date], settings: Settings ) -> tuple[Report | None, str | None, str | None]: """Return (report, model_used, error); any failure degrades instead of aborting (pipeline policy).""" try: provider = build_provider(settings) report, model_used = summarize_weekly_payload(payload, sessions, provider) return report, model_used, None except LLMError as exc: log.error("weekly LLM summarisation failed, degrading: %s", exc) return None, None, str(exc) except Exception as exc: log.exception("unexpected weekly LLM failure, degrading") return None, None, f"{type(exc).__name__}: {exc}" def _verify_weekly( payload: MarketPayload, sessions: list[date], report: Report, settings: Settings ) -> tuple[Report, Verification | None]: try: provider = build_provider(settings) return verify_and_revise( payload, report, provider, data_message=build_weekly_user_message(payload, sessions), system_prompt=WEEKLY_SYSTEM_PROMPT, ) except Exception as exc: # noqa: BLE001 - verification must never block the recap log.warning("weekly verification failed, delivering unverified: %s: %s", type(exc).__name__, exc) return report, None def run_weekly( monday: date | None, settings: Settings, *, dry_run: bool = False, force: bool = False, no_llm: bool = False, ) -> BriefingResult: """Produce and deliver the weekly recap for ``monday`` (default: the most recent Monday, KST).""" clock = MarketClock(settings.run_offset_minutes, extra_krx_holidays=settings.krx_extra_holidays) store = Store(settings.db_path) try: now = _now() if monday is None: monday = most_recent_monday(now.date()) if monday.weekday() != 0: raise ValueError(f"{monday} is not a Monday - the weekly recap is keyed by Mondays (KST)") if monday > now.date(): raise ValueError(f"{monday} is in the future") payload, sessions = build_weekly_payload(monday, settings, clock, store) if not dry_run and not force and store.has_report(monday, kind="weekly"): log.info("weekly recap for %s already sent - skipping (use --force to resend)", monday) return _skip_result(payload, SKIP_ALREADY_SENT) 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 weekly summarisation") else: report, model_used, error = _summarize_weekly(payload, sessions, settings) degraded = report is None if report is not None and settings.verify_enabled: report, verification = _verify_weekly(payload, sessions, report, settings) markdown = render_weekly_markdown(payload, report, sessions) result = BriefingResult( payload=payload, report=report, markdown=markdown, model_used=model_used, degraded=degraded, error=error, verification=verification, kind="weekly", ) if dry_run: print(markdown) return result report_id = store.save_report(result, kind="weekly") log.info("weekly report %d saved for %s (degraded=%s)", report_id, monday, degraded) _deliver(result, settings, store, report_id) return result finally: store.close()