You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
100 lines
3.3 KiB
Python
100 lines
3.3 KiB
Python
"""Parquet snapshot store for Yahoo day_gainers ticks.
|
|
|
|
Layout: data/cache/yahoo_gainers/{YYYY-MM-DD}/{HH-MM-SS}.parquet
|
|
|
|
Each file contains one tick's worth of GainerQuote rows.
|
|
The SQLite tgtc_snapshots table mirrors this for fast UI polling.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
_DEFAULT_CACHE_DIR = "data/cache/yahoo_gainers"
|
|
|
|
|
|
def save_snapshot_parquet(
|
|
quotes: list, # list[GainerQuote]
|
|
date_str: str,
|
|
captured_at_iso: str,
|
|
cache_dir: str = _DEFAULT_CACHE_DIR,
|
|
) -> None:
|
|
"""Write one snapshot tick to Parquet (atomic)."""
|
|
try:
|
|
import pyarrow as pa
|
|
import pyarrow.parquet as pq
|
|
except ImportError:
|
|
log.warning("TGTC snapshot_store: pyarrow not available, skipping Parquet write")
|
|
return
|
|
|
|
if not quotes:
|
|
return
|
|
|
|
# captured_at_iso like "2026-05-05T09:35:00"
|
|
safe_ts = captured_at_iso.replace(":", "-").replace("T", "_")[:16] # "2026-05-05_09-35"
|
|
dir_path = Path(cache_dir) / date_str
|
|
dir_path.mkdir(parents=True, exist_ok=True)
|
|
file_path = dir_path / f"{safe_ts}.parquet"
|
|
|
|
schema = pa.schema([
|
|
pa.field("captured_at", pa.string()),
|
|
pa.field("symbol", pa.string()),
|
|
pa.field("rank", pa.int32()),
|
|
pa.field("price", pa.float64()),
|
|
pa.field("pct_change", pa.float64()),
|
|
pa.field("volume", pa.float64()),
|
|
pa.field("market_cap", pa.float64()),
|
|
])
|
|
|
|
table = pa.table(
|
|
{
|
|
"captured_at": [captured_at_iso] * len(quotes),
|
|
"symbol": [q.symbol for q in quotes],
|
|
"rank": [q.rank for q in quotes],
|
|
"price": [q.price for q in quotes],
|
|
"pct_change": [q.pct_change for q in quotes],
|
|
"volume": [q.volume for q in quotes],
|
|
"market_cap": [q.market_cap for q in quotes],
|
|
},
|
|
schema=schema,
|
|
)
|
|
|
|
# Atomic write via temp file
|
|
tmp_path = file_path.with_suffix(f".{os.urandom(4).hex()}.tmp")
|
|
try:
|
|
pq.write_table(table, str(tmp_path), compression="snappy")
|
|
os.replace(str(tmp_path), str(file_path))
|
|
except Exception as exc:
|
|
log.error("TGTC snapshot_store: write failed %s: %s", file_path, exc)
|
|
if tmp_path.exists():
|
|
tmp_path.unlink(missing_ok=True)
|
|
|
|
|
|
def load_snapshots_for_date(date_str: str,
|
|
cache_dir: str = _DEFAULT_CACHE_DIR) -> list[dict]:
|
|
"""Load all snapshot ticks for a date from Parquet files, sorted by captured_at."""
|
|
try:
|
|
import pyarrow.parquet # noqa: F401
|
|
except ImportError:
|
|
return []
|
|
|
|
dir_path = Path(cache_dir) / date_str
|
|
if not dir_path.exists():
|
|
return []
|
|
|
|
rows: list[dict] = []
|
|
for parquet_file in sorted(dir_path.glob("*.parquet")):
|
|
try:
|
|
import pyarrow.parquet as pq
|
|
table = pq.read_table(str(parquet_file))
|
|
for batch in table.to_batches():
|
|
rows.extend(dict(zip(table.schema.names, row))
|
|
for row in zip(*[batch.column(i).to_pylist()
|
|
for i in range(len(table.schema))]))
|
|
except Exception as exc:
|
|
log.warning("TGTC snapshot_store: failed to read %s: %s", parquet_file, exc)
|
|
return rows
|