File size: 5,250 Bytes
559c2ff d1e793b 559c2ff d1e793b 559c2ff 8dbbf97 e6496c0 559c2ff 8dbbf97 559c2ff 8dbbf97 559c2ff 8dbbf97 559c2ff 8dbbf97 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 | """storage/sections_db.py — persist raw section text across filing periods.
Stores verbatim Business, Segments/Geography, MD&A, Risk Factors, and
transcript text keyed by (ticker, period, section). Used by company-profile
generation and analysis/textdiff.py without re-fetching source documents.
This is append-only at ingest time; textdiff reads it at runtime.
"""
from __future__ import annotations
import sqlite3
from pathlib import Path
from typing import Optional
SECTIONS_DB_PATH = Path("data/sections.db")
_SCHEMA = """
CREATE TABLE IF NOT EXISTS sections (
ticker TEXT NOT NULL,
period TEXT NOT NULL,
form_type TEXT NOT NULL,
section TEXT NOT NULL,
text TEXT NOT NULL DEFAULT '',
ingested_at TEXT,
PRIMARY KEY (ticker, period, section)
)
"""
def init_sections_db() -> None:
SECTIONS_DB_PATH.parent.mkdir(parents=True, exist_ok=True)
with sqlite3.connect(SECTIONS_DB_PATH) as conn:
conn.execute(_SCHEMA)
def upsert_section(
ticker: str,
period: str,
form_type: str,
section: str,
text: str,
) -> None:
"""Write or overwrite a supported filing or transcript section."""
from datetime import datetime, timezone
SECTIONS_DB_PATH.parent.mkdir(parents=True, exist_ok=True)
with sqlite3.connect(SECTIONS_DB_PATH) as conn:
conn.execute(_SCHEMA)
conn.execute(
"""
INSERT INTO sections (ticker, period, form_type, section, text, ingested_at)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(ticker, period, section) DO UPDATE SET
form_type = excluded.form_type,
text = excluded.text,
ingested_at = excluded.ingested_at
""",
(
ticker.upper(),
period,
form_type,
section,
text or "",
datetime.now(timezone.utc).isoformat(),
),
)
def get_section(ticker: str, period: str, section: str) -> Optional[str]:
"""Return the stored text for (ticker, period, section), or None if absent."""
if not SECTIONS_DB_PATH.exists():
return None
with sqlite3.connect(SECTIONS_DB_PATH) as conn:
row = conn.execute(
"SELECT text FROM sections WHERE ticker=? AND period=? AND section=?",
(ticker.upper(), period, section),
).fetchone()
return row[0] if row else None
def _period_sort_key(period: str) -> tuple[int, int]:
"""Parse 'Q12027' → (2027, 1) for correct chronological sort (newest first).
Falls back to (0, 0) for unparseable strings (e.g. 'FY2024').
"""
if not period:
return (0, 0)
if period.startswith("Q") and len(period) >= 6:
try:
body = period[1:] # "12027"
year = int(body[-4:]) # 2027
quarter = int(body[:-4]) # 1
return (year, quarter)
except (ValueError, IndexError):
pass
if period.startswith("FY") and len(period) == 6:
try:
return (int(period[2:]), 0)
except ValueError:
pass
return (0, 0)
def _transcript_sort_key(period: str) -> tuple[int, int]:
"""Chronological sort key that places FY calls as Q4 of their year.
'Q12026' → (2026, 1); 'FY2025' → (2025, 4) so the Q4/FY earnings call
sorts between Q32025 and Q12026. Unparseable strings → (0, 0).
"""
if period and period.startswith("FY") and len(period) == 6:
try:
return (int(period[2:]), 4)
except ValueError:
return (0, 0)
return _period_sort_key(period)
def get_recent_transcripts(ticker: str, n: int = 4) -> list[tuple[str, str]]:
"""Return [(period, text)] for the n most recent non-empty transcripts.
Mixes 10-Q and 10-K (FY) earnings calls; FY periods sort as Q4 of their
year. Results are in chronological order (oldest first).
"""
if not SECTIONS_DB_PATH.exists():
return []
with sqlite3.connect(SECTIONS_DB_PATH) as conn:
rows = conn.execute(
"SELECT period, text FROM sections "
"WHERE ticker=? AND section='transcript' AND length(text) > 0",
(ticker.upper(),),
).fetchall()
rows.sort(key=lambda r: _transcript_sort_key(r[0]))
return [(r[0], r[1]) for r in rows[-n:]]
def get_periods_for_ticker(ticker: str, form_type: Optional[str] = None) -> list[str]:
"""Return all period strings stored for a ticker, sorted newest first (chronologically).
Optionally filtered by form_type (e.g. '10-Q').
"""
if not SECTIONS_DB_PATH.exists():
return []
with sqlite3.connect(SECTIONS_DB_PATH) as conn:
if form_type:
rows = conn.execute(
"SELECT DISTINCT period FROM sections WHERE ticker=? AND form_type=?",
(ticker.upper(), form_type),
).fetchall()
else:
rows = conn.execute(
"SELECT DISTINCT period FROM sections WHERE ticker=?",
(ticker.upper(),),
).fetchall()
periods = [r[0] for r in rows]
periods.sort(key=_period_sort_key, reverse=True)
return periods
|