419 lines
16 KiB
Python
419 lines
16 KiB
Python
"""SQLite persistence: quote history, report idempotency, delivery receipts, raw collector payloads.
|
|
|
|
Single-threaded use only (one connection, WAL mode). Dates are stored as ISO ``YYYY-MM-DD`` text so
|
|
lexicographic ordering equals chronological ordering.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import math
|
|
import sqlite3
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, date, datetime
|
|
from pathlib import Path
|
|
from typing import Self
|
|
|
|
from us2kr.models import BriefingResult, Quote
|
|
from us2kr.symbols import POINT_CHANGE_GROUPS, display_name, group_of
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
# Symbols for which history-based notes are generated (order = order of notes in the briefing).
|
|
HISTORY_SYMBOLS: tuple[str, ...] = ("^GSPC", "^IXIC", "^SOX", "^VIX", "KRW=X", "^TNX", "NVDA", "EWY")
|
|
HISTORY_SESSIONS = 10 # look back at most this many stored sessions
|
|
MIN_STREAK_DAYS = 3
|
|
MIN_SESSIONS_FOR_AVG = 5 # need at least this many prior sessions to judge "unusual" moves
|
|
BIG_MOVE_RATIO = 2.0
|
|
VIX_LEVELS: tuple[float, ...] = (20.0, 30.0)
|
|
KRW_STEP = 50.0
|
|
# Streak contiguity: two stored rows count as consecutive sessions only when the calendar gap is at most
|
|
# a weekend plus one holiday (Fri -> Tue) AND the later row's prev_close matches the earlier row's close
|
|
# (yfinance revisions can shift a close slightly, hence the relative tolerance).
|
|
MAX_SESSION_GAP_DAYS = 4
|
|
PREV_CLOSE_TOLERANCE = 0.002
|
|
|
|
_SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS quotes (
|
|
session_date TEXT NOT NULL,
|
|
symbol TEXT NOT NULL,
|
|
close REAL,
|
|
prev_close REAL,
|
|
change_pct REAL,
|
|
change_abs REAL,
|
|
day_high REAL,
|
|
day_low REAL,
|
|
volume REAL,
|
|
PRIMARY KEY (session_date, symbol)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS reports (
|
|
id INTEGER PRIMARY KEY,
|
|
krx_date TEXT NOT NULL,
|
|
kind TEXT NOT NULL,
|
|
session_date TEXT,
|
|
model_used TEXT,
|
|
degraded INTEGER NOT NULL DEFAULT 0,
|
|
markdown TEXT,
|
|
result_json TEXT,
|
|
created_at TEXT,
|
|
UNIQUE (krx_date, kind)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS deliveries (
|
|
id INTEGER PRIMARY KEY,
|
|
report_id INTEGER NOT NULL,
|
|
channel TEXT NOT NULL,
|
|
receipt TEXT,
|
|
sent_at TEXT
|
|
);
|
|
CREATE TABLE IF NOT EXISTS raw (
|
|
session_date TEXT NOT NULL,
|
|
name TEXT NOT NULL,
|
|
payload_json TEXT,
|
|
created_at TEXT,
|
|
PRIMARY KEY (session_date, name)
|
|
);
|
|
"""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _Row:
|
|
"""One stored daily quote (subset of columns needed for history notes)."""
|
|
|
|
session_date: str
|
|
close: float | None
|
|
prev_close: float | None
|
|
change_pct: float | None
|
|
change_abs: float | None
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(UTC).isoformat(timespec="seconds")
|
|
|
|
|
|
class Store:
|
|
"""Thin sqlite3 wrapper. Creates parent directories and tables on construction."""
|
|
|
|
def __init__(self, db_path: Path) -> None:
|
|
self.db_path = Path(db_path)
|
|
self.db_path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._conn = sqlite3.connect(str(self.db_path), timeout=30, isolation_level=None)
|
|
self._conn.row_factory = sqlite3.Row
|
|
self._conn.execute("PRAGMA journal_mode=WAL")
|
|
self._conn.execute("PRAGMA synchronous=NORMAL")
|
|
self._conn.executescript(_SCHEMA)
|
|
log.debug("store opened at %s", self.db_path)
|
|
|
|
# ------------------------------------------------------------------ quotes
|
|
def save_quotes(self, session_date: date, quotes: list[Quote]) -> None:
|
|
"""Upsert the day's quotes keyed by (session_date, symbol)."""
|
|
rows = [
|
|
(
|
|
session_date.isoformat(),
|
|
q.symbol,
|
|
q.close,
|
|
q.prev_close,
|
|
q.change_pct,
|
|
q.change_abs,
|
|
q.day_high,
|
|
q.day_low,
|
|
q.volume,
|
|
)
|
|
for q in quotes
|
|
]
|
|
with self._conn:
|
|
self._conn.executemany(
|
|
"""
|
|
INSERT INTO quotes (session_date, symbol, close, prev_close, change_pct, change_abs,
|
|
day_high, day_low, volume)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(session_date, symbol) DO UPDATE SET
|
|
close = excluded.close,
|
|
prev_close = excluded.prev_close,
|
|
change_pct = excluded.change_pct,
|
|
change_abs = excluded.change_abs,
|
|
day_high = excluded.day_high,
|
|
day_low = excluded.day_low,
|
|
volume = excluded.volume
|
|
""",
|
|
rows,
|
|
)
|
|
log.info("saved %d quotes for %s", len(rows), session_date)
|
|
|
|
def load_quotes(self, session_date: date) -> list[sqlite3.Row]:
|
|
"""Return stored quote rows for one session (mainly for tests / debugging)."""
|
|
cur = self._conn.execute(
|
|
"SELECT * FROM quotes WHERE session_date = ? ORDER BY symbol", (session_date.isoformat(),)
|
|
)
|
|
return list(cur.fetchall())
|
|
|
|
def _history(self, symbol: str, before: date, n: int) -> list[_Row]:
|
|
"""Up to ``n`` stored rows for ``symbol`` strictly before ``before``, oldest first."""
|
|
cur = self._conn.execute(
|
|
"""
|
|
SELECT session_date, close, prev_close, change_pct, change_abs
|
|
FROM quotes WHERE symbol = ? AND session_date < ?
|
|
ORDER BY session_date DESC LIMIT ?
|
|
""",
|
|
(symbol, before.isoformat(), n),
|
|
)
|
|
rows = [
|
|
_Row(r["session_date"], r["close"], r["prev_close"], r["change_pct"], r["change_abs"])
|
|
for r in cur
|
|
]
|
|
rows.reverse()
|
|
return rows
|
|
|
|
def history_notes(self, session_date: date, quotes: list[Quote]) -> list[str]:
|
|
"""Korean one-liners derived from stored history + today's quotes.
|
|
|
|
Covers: N-day streaks (N>=3), today's move vs the 10-session mean absolute move,
|
|
VIX crossing 20/30 and USD/KRW crossing a multiple of 50 won. Returns [] without history.
|
|
"""
|
|
today = {q.symbol: q for q in quotes}
|
|
notes: list[str] = []
|
|
for sym in HISTORY_SYMBOLS:
|
|
q = today.get(sym)
|
|
if q is None or q.close is None:
|
|
continue
|
|
hist = self._history(sym, session_date, HISTORY_SESSIONS)
|
|
if not hist:
|
|
continue
|
|
name = display_name(sym)
|
|
points = (q.group or group_of(sym)) in POINT_CHANGE_GROUPS
|
|
today_delta = _delta(q.change_pct, q.change_abs, q.close, q.prev_close, points)
|
|
deltas = [_delta(r.change_pct, r.change_abs, r.close, r.prev_close, points) for r in hist]
|
|
|
|
# Streaks may only count sessions forming an unbroken chain ending today; a hole in the table
|
|
# (daemon down past the catch-up window, prices collector failed) must not read as "N일 연속".
|
|
chain = _contiguous_tail(hist, session_date, q.prev_close)
|
|
chain_deltas = [_delta(r.change_pct, r.change_abs, r.close, r.prev_close, points) for r in chain]
|
|
note = _streak_note(name, chain_deltas + [today_delta])
|
|
if note:
|
|
notes.append(note)
|
|
note = _big_move_note(name, deltas, today_delta, points) # a sample mean: holes are harmless
|
|
if note:
|
|
notes.append(note)
|
|
|
|
# Level crossings must agree with the rendered table, which is built from q.prev_close (the
|
|
# collector's directly preceding bar). Stored history is only a fallback: it may be a still-live
|
|
# snapshot for 24h instruments (KRW=X) or 2+ sessions old after a missed run.
|
|
prev_close = q.prev_close if q.prev_close is not None else hist[-1].close
|
|
if prev_close is None:
|
|
continue
|
|
if sym == "^VIX":
|
|
notes.extend(_level_cross_notes(name, prev_close, q.close, list(VIX_LEVELS), 1))
|
|
elif sym == "KRW=X":
|
|
notes.extend(
|
|
_level_cross_notes(name, prev_close, q.close, _krw_levels(prev_close, q.close), 0, "원")
|
|
)
|
|
return notes
|
|
|
|
# ----------------------------------------------------------------- reports
|
|
def has_report(self, krx_date: date, kind: str = "briefing") -> bool:
|
|
cur = self._conn.execute(
|
|
"SELECT 1 FROM reports WHERE krx_date = ? AND kind = ? LIMIT 1", (krx_date.isoformat(), kind)
|
|
)
|
|
return cur.fetchone() is not None
|
|
|
|
def save_report(self, result: BriefingResult, kind: str = "briefing") -> int:
|
|
"""Upsert the briefing on (krx_date, kind); return the report id."""
|
|
krx = result.payload.krx_date.isoformat()
|
|
with self._conn:
|
|
self._conn.execute(
|
|
"""
|
|
INSERT INTO reports (krx_date, kind, session_date, model_used, degraded, markdown,
|
|
result_json, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(krx_date, kind) DO UPDATE SET
|
|
session_date = excluded.session_date,
|
|
model_used = excluded.model_used,
|
|
degraded = excluded.degraded,
|
|
markdown = excluded.markdown,
|
|
result_json = excluded.result_json,
|
|
created_at = excluded.created_at
|
|
""",
|
|
(
|
|
krx,
|
|
kind,
|
|
result.payload.session_date.isoformat(),
|
|
result.model_used,
|
|
int(result.degraded),
|
|
result.markdown,
|
|
result.model_dump_json(),
|
|
_now_iso(),
|
|
),
|
|
)
|
|
row = self._conn.execute(
|
|
"SELECT id FROM reports WHERE krx_date = ? AND kind = ?", (krx, kind)
|
|
).fetchone()
|
|
report_id = int(row["id"])
|
|
log.info("saved report id=%d krx_date=%s kind=%s degraded=%s", report_id, krx, kind, result.degraded)
|
|
return report_id
|
|
|
|
def mark_sent(self, report_id: int, channel: str, receipt: str) -> None:
|
|
with self._conn:
|
|
self._conn.execute(
|
|
"INSERT INTO deliveries (report_id, channel, receipt, sent_at) VALUES (?, ?, ?, ?)",
|
|
(report_id, channel, receipt, _now_iso()),
|
|
)
|
|
|
|
def deliveries_for(self, report_id: int) -> list[sqlite3.Row]:
|
|
cur = self._conn.execute("SELECT * FROM deliveries WHERE report_id = ? ORDER BY id", (report_id,))
|
|
return list(cur.fetchall())
|
|
|
|
# --------------------------------------------------------------------- raw
|
|
def save_raw(self, session_date: date, name: str, payload_json: str) -> None:
|
|
with self._conn:
|
|
self._conn.execute(
|
|
"""
|
|
INSERT INTO raw (session_date, name, payload_json, created_at) VALUES (?, ?, ?, ?)
|
|
ON CONFLICT(session_date, name) DO UPDATE SET
|
|
payload_json = excluded.payload_json, created_at = excluded.created_at
|
|
""",
|
|
(session_date.isoformat(), name, payload_json, _now_iso()),
|
|
)
|
|
|
|
def load_raw(self, session_date: date, name: str) -> str | None:
|
|
row = self._conn.execute(
|
|
"SELECT payload_json FROM raw WHERE session_date = ? AND name = ?",
|
|
(session_date.isoformat(), name),
|
|
).fetchone()
|
|
return None if row is None else row["payload_json"]
|
|
|
|
# ------------------------------------------------------------------- misc
|
|
def close(self) -> None:
|
|
try:
|
|
self._conn.close()
|
|
except sqlite3.Error: # pragma: no cover - defensive
|
|
log.debug("store close failed", exc_info=True)
|
|
|
|
def __enter__(self) -> Self:
|
|
return self
|
|
|
|
def __exit__(self, *exc: object) -> None:
|
|
self.close()
|
|
|
|
|
|
# ----------------------------------------------------------------------- helpers
|
|
|
|
|
|
def _delta(
|
|
change_pct: float | None,
|
|
change_abs: float | None,
|
|
close: float | None,
|
|
prev_close: float | None,
|
|
points: bool,
|
|
) -> float | None:
|
|
"""Daily move in the unit relevant for the symbol: yield points for rates, percent otherwise."""
|
|
if points:
|
|
if change_abs is not None:
|
|
return change_abs
|
|
if close is not None and prev_close is not None:
|
|
return close - prev_close
|
|
return None
|
|
if change_pct is not None:
|
|
return change_pct
|
|
if close is not None and prev_close:
|
|
return (close / prev_close - 1.0) * 100.0
|
|
return None
|
|
|
|
|
|
def _is_predecessor(older: _Row, newer_date: str, newer_prev_close: float | None) -> bool:
|
|
"""True when ``older`` looks like the trading session immediately before ``newer_date``."""
|
|
gap = (date.fromisoformat(newer_date) - date.fromisoformat(older.session_date)).days
|
|
if gap <= 0 or gap > MAX_SESSION_GAP_DAYS:
|
|
return False
|
|
if older.close and newer_prev_close is not None:
|
|
return abs(newer_prev_close / older.close - 1.0) <= PREV_CLOSE_TOLERANCE
|
|
return True # no prices to cross-check; the calendar gap is plausible
|
|
|
|
|
|
def _contiguous_tail(hist: list[_Row], session_date: date, today_prev_close: float | None) -> list[_Row]:
|
|
"""Longest suffix of ``hist`` (oldest first) forming an unbroken session chain ending at today."""
|
|
tail: list[_Row] = []
|
|
newer_date, newer_prev = session_date.isoformat(), today_prev_close
|
|
for r in reversed(hist):
|
|
if not _is_predecessor(r, newer_date, newer_prev):
|
|
break
|
|
tail.append(r)
|
|
newer_date, newer_prev = r.session_date, r.prev_close
|
|
tail.reverse()
|
|
return tail
|
|
|
|
|
|
def _streak_note(name: str, deltas: list[float | None]) -> str | None:
|
|
"""'{name} N일 연속 상승/하락' when the last N>=3 deltas share a sign (last element = today)."""
|
|
if not deltas or deltas[-1] is None or deltas[-1] == 0:
|
|
return None
|
|
up = deltas[-1] > 0
|
|
n = 0
|
|
for d in reversed(deltas):
|
|
if d is None or d == 0 or (d > 0) != up:
|
|
break
|
|
n += 1
|
|
if n < MIN_STREAK_DAYS:
|
|
return None
|
|
return f"{name} {n}일 연속 {'상승' if up else '하락'}"
|
|
|
|
|
|
def _big_move_note(
|
|
name: str, hist_deltas: list[float | None], today: float | None, points: bool
|
|
) -> str | None:
|
|
"""Flag today's move when it exceeds BIG_MOVE_RATIO x the mean absolute move of stored sessions."""
|
|
moves = [abs(d) for d in hist_deltas if d is not None]
|
|
if today is None or len(moves) < MIN_SESSIONS_FOR_AVG:
|
|
return None
|
|
mean = sum(moves) / len(moves)
|
|
if mean <= 0 or abs(today) <= BIG_MOVE_RATIO * mean:
|
|
return None
|
|
ratio = abs(today) / mean
|
|
if points:
|
|
return (
|
|
f"{name} 오늘 {today * 100:+.1f}bp — 최근 {len(moves)}일 평균 변동폭({mean * 100:.1f}bp)의 "
|
|
f"{ratio:.1f}배, 평소보다 큰 변동"
|
|
)
|
|
return f"{name} 오늘 {today:+.2f}% — 최근 {len(moves)}일 평균 변동폭({mean:.2f}%)의 {ratio:.1f}배, 평소보다 큰 변동"
|
|
|
|
|
|
def _krw_levels(prev: float, close: float) -> list[float]:
|
|
"""Candidate multiples of KRW_STEP spanning [prev, close]; crossing is decided by the caller."""
|
|
lo, hi = sorted((prev, close))
|
|
first = math.floor(lo / KRW_STEP) * KRW_STEP
|
|
last = math.ceil(hi / KRW_STEP) * KRW_STEP
|
|
return [first + i * KRW_STEP for i in range(round((last - first) / KRW_STEP) + 1)]
|
|
|
|
|
|
def _fmt_pair(a: float, b: float, decimals: int) -> tuple[str, str]:
|
|
"""Format two values, adding precision (up to 4 dp) until the strings differ.
|
|
|
|
Otherwise a 1399.99 -> 1400.0 break would read "1,400 → 1,400", contradicting itself.
|
|
"""
|
|
sa, sb = f"{a:,.{decimals}f}", f"{b:,.{decimals}f}"
|
|
for d in range(decimals + 1, 5):
|
|
if sa != sb:
|
|
break
|
|
sa, sb = f"{a:,.{d}f}", f"{b:,.{d}f}"
|
|
return sa, sb
|
|
|
|
|
|
def _level_cross_notes(
|
|
name: str, prev: float, close: float, levels: list[float], decimals: int, unit: str = ""
|
|
) -> list[str]:
|
|
"""Note for a level strictly crossed between prev and close.
|
|
|
|
Up: prev < L < close; down: prev > L > close.
|
|
|
|
Touching a level exactly (landing on it or leaving from it) is not a break.
|
|
"""
|
|
notes: list[str] = []
|
|
crossed = [lvl for lvl in levels if (prev < lvl < close) or (prev > lvl > close)]
|
|
if not crossed:
|
|
return notes
|
|
# report only the level nearest to today's close (avoid spam if several were crossed)
|
|
lvl = min(crossed, key=lambda x: abs(x - close))
|
|
direction = "상향 돌파" if close > prev else "하향 이탈"
|
|
lvl_txt = f"{lvl:,.0f}{unit}"
|
|
prev_txt, close_txt = _fmt_pair(prev, close, decimals)
|
|
notes.append(f"{name} {lvl_txt} {direction} ({prev_txt} → {close_txt})")
|
|
return notes
|