Gahow Wang
838749d8b2
The append-only updater only fetched when last_date+1 < local date, never revisited appended rows and forward-filled before saving. First-seen values (partial intraday bars, late or missing prints) were frozen permanently: vs Yahoo, 22% (US), 58% (CN) and 27% (ETF) of recent days had >0.5% daily-return errors, CN never received same-day data, and HK tickers created US-holiday rows. Panels are now stored raw (NaN = no print) and forward-filled on read; every update re-downloads the last 30 days for the requested tickers, rescales older history when Yahoo's split/dividend adjustment moves, drops rows printed by a minority of tickers, and writes atomically under a per-market file lock. Tickers Yahoo has no data for are retried weekly instead of every run. rebuild() repairs panels written by the old updater. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
384 lines
16 KiB
Python
384 lines
16 KiB
Python
"""
|
|
Persistent local data store for price history.
|
|
|
|
Layout:
|
|
data/{market}.csv — adjusted close prices
|
|
data/{market}_open.csv — adjusted open prices
|
|
data/{market}_unavailable.json — tickers Yahoo returned no data for (retried weekly)
|
|
|
|
Panels are stored raw: a date/ticker Yahoo has no bar for stays NaN, so a
|
|
missing print is distinguishable from an unchanged price. ``load()`` and
|
|
``update()`` forward-fill on read; ``load_raw()`` exposes the stored values.
|
|
|
|
Every ``update()`` re-downloads a trailing ``REFRESH_DAYS`` window and
|
|
overwrites it for the requested tickers. Partial intraday bars, late prints and
|
|
rows that turn out not to be trading days are therefore corrected on the next
|
|
run instead of being frozen. When Yahoo's split/dividend adjustment shifts the
|
|
overlap, that ticker's older history is rescaled so returns stay continuous.
|
|
"""
|
|
|
|
import fcntl
|
|
import json
|
|
import os
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timedelta
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import yfinance as yf
|
|
|
|
DATA_DIR = "data"
|
|
|
|
# Trailing calendar days re-downloaded and overwritten on every update.
|
|
REFRESH_DAYS = 30
|
|
# A row is kept only if at least this fraction of tickers alive on that date
|
|
# have a bar. Drops e.g. US-holiday rows created by HK-listed tickers.
|
|
MIN_ROW_COVERAGE = 0.5
|
|
# Relative change in the adjusted-price overlap that triggers a rescale.
|
|
RESCALE_TOLERANCE = 1e-3
|
|
UNAVAILABLE_RETRY_DAYS = 7
|
|
|
|
|
|
def _data_path(market: str, price_type: str = "close") -> str:
|
|
suffix = "" if price_type == "close" else f"_{price_type}"
|
|
return os.path.join(DATA_DIR, f"{market}{suffix}.csv")
|
|
|
|
|
|
def load_raw(market: str, price_type: str = "close") -> pd.DataFrame | None:
|
|
"""Load the stored panel without filling missing bars, or None if absent."""
|
|
path = _data_path(market, price_type)
|
|
if not os.path.exists(path):
|
|
return None
|
|
return pd.read_csv(path, index_col=0, parse_dates=True)
|
|
|
|
|
|
def load(market: str, price_type: str = "close") -> pd.DataFrame | None:
|
|
"""Load the stored panel with missing bars forward-filled."""
|
|
raw = load_raw(market, price_type)
|
|
return None if raw is None else raw.ffill()
|
|
|
|
|
|
def _write_atomic(df: pd.DataFrame, path: str) -> None:
|
|
tmp = f"{path}.tmp"
|
|
df.to_csv(tmp)
|
|
os.replace(tmp, path)
|
|
|
|
|
|
@contextmanager
|
|
def _panel_lock(market: str):
|
|
"""Serialize updates of one market panel across processes (monitor, bridge, scripts)."""
|
|
os.makedirs(DATA_DIR, exist_ok=True)
|
|
with open(os.path.join(DATA_DIR, f".{market}.lock"), "w") as fh:
|
|
fcntl.flock(fh, fcntl.LOCK_EX)
|
|
try:
|
|
yield
|
|
finally:
|
|
fcntl.flock(fh, fcntl.LOCK_UN)
|
|
|
|
|
|
def _download(tickers: list[str], start: str, end: str | None = None,
|
|
fields: list[str] | None = None) -> dict[str, pd.DataFrame]:
|
|
"""
|
|
Download price data from Yahoo Finance.
|
|
|
|
Parameters
|
|
----------
|
|
fields : list of str, optional
|
|
Price fields to extract, e.g. ["Close", "Open"].
|
|
Defaults to ["Close"].
|
|
|
|
Returns
|
|
-------
|
|
dict mapping field name to DataFrame (columns = tickers).
|
|
"""
|
|
if fields is None:
|
|
fields = ["Close"]
|
|
print(f"--- Downloading {len(tickers)} tickers from {start}{f' to {end}' if end else ''} ---")
|
|
kwargs = dict(auto_adjust=True)
|
|
if end:
|
|
kwargs["end"] = end
|
|
raw = yf.download(tickers, start=start, **kwargs)
|
|
if raw.empty:
|
|
return {f: pd.DataFrame() for f in fields}
|
|
|
|
result = {}
|
|
for field in fields:
|
|
if field in raw.columns.get_level_values(0) if isinstance(raw.columns, pd.MultiIndex) else field in raw.columns:
|
|
selected = raw[field]
|
|
if isinstance(selected, pd.Series):
|
|
result[field] = selected.to_frame(name=tickers[0])
|
|
else:
|
|
result[field] = selected
|
|
else:
|
|
result[field] = pd.DataFrame()
|
|
return result
|
|
|
|
|
|
def _clean_market_data(data: pd.DataFrame, field: str) -> pd.DataFrame:
|
|
"""Clean market data while preserving volume gaps."""
|
|
good = data.columns[data.notna().mean() > 0.5]
|
|
dropped = set(data.columns) - set(good)
|
|
if dropped:
|
|
print(f"--- Dropped {len(dropped)} tickers with >50% missing data ---")
|
|
data = data[good]
|
|
if field == "volume":
|
|
return data
|
|
return data.ffill().dropna(how="all")
|
|
|
|
|
|
def _merge_market_panel(existing: pd.DataFrame | None, new_data: pd.DataFrame) -> pd.DataFrame:
|
|
"""Merge new data into an existing cached panel, preserving old columns and dates."""
|
|
if existing is None or existing.empty:
|
|
merged = new_data.copy()
|
|
elif new_data.empty:
|
|
merged = existing.copy()
|
|
else:
|
|
merged = existing.combine_first(new_data)
|
|
merged.loc[new_data.index, new_data.columns] = new_data
|
|
merged = merged.sort_index()
|
|
merged = merged[~merged.index.duplicated(keep="last")]
|
|
return merged
|
|
|
|
|
|
def update_market_data(market: str, tickers: list[str], fields: list[str]) -> dict[str, pd.DataFrame]:
|
|
"""Download, clean, persist, and return market data panels for requested Yahoo fields."""
|
|
field_aliases = {
|
|
"close": "Close",
|
|
"open": "Open",
|
|
"high": "High",
|
|
"low": "Low",
|
|
"volume": "Volume",
|
|
}
|
|
normalized_fields = []
|
|
yahoo_fields = []
|
|
for field in fields:
|
|
normalized = field.lower()
|
|
if normalized not in field_aliases:
|
|
raise ValueError(f"Unsupported market data field: {field}")
|
|
normalized_fields.append(normalized)
|
|
yahoo_fields.append(field_aliases[normalized])
|
|
|
|
os.makedirs(DATA_DIR, exist_ok=True)
|
|
start = (datetime.now() - timedelta(days=365 * 10)).strftime("%Y-%m-%d")
|
|
downloaded = _download(tickers, start=start, fields=yahoo_fields)
|
|
|
|
cleaned = {}
|
|
for normalized, yahoo_field in zip(normalized_fields, yahoo_fields):
|
|
data = _clean_market_data(downloaded.get(yahoo_field, pd.DataFrame()), normalized)
|
|
existing = load(market, normalized)
|
|
data = _merge_market_panel(existing, data)
|
|
path = _data_path(market, normalized)
|
|
data.to_csv(path)
|
|
print(f"--- Saved {data.shape[0]} days x {data.shape[1]} tickers to {path} ---")
|
|
cleaned[normalized] = data
|
|
return cleaned
|
|
|
|
|
|
|
|
def _unavailable_path(market: str) -> str:
|
|
return os.path.join(DATA_DIR, f"{market}_unavailable.json")
|
|
|
|
|
|
def _load_unavailable(market: str) -> dict[str, str]:
|
|
try:
|
|
with open(_unavailable_path(market)) as f:
|
|
return json.load(f)
|
|
except (OSError, ValueError):
|
|
return {}
|
|
|
|
|
|
def _save_unavailable(market: str, unavailable: dict[str, str]) -> None:
|
|
with open(_unavailable_path(market), "w") as f:
|
|
json.dump(unavailable, f, indent=0, sort_keys=True)
|
|
|
|
|
|
def _drop_sparse_rows(df: pd.DataFrame) -> pd.DataFrame:
|
|
"""Drop rows where too few of the tickers alive on that date have a bar."""
|
|
df = df.dropna(how="all")
|
|
if df.empty:
|
|
return df
|
|
notna = df.notna().to_numpy()
|
|
n = len(df)
|
|
has_any = notna.any(axis=0)
|
|
first = np.where(has_any, notna.argmax(axis=0), n)
|
|
last = np.where(has_any, n - 1 - notna[::-1].argmax(axis=0), -1)
|
|
pos = np.arange(n)[:, None]
|
|
alive = (pos >= first) & (pos <= last)
|
|
coverage = notna.sum(axis=1) / np.maximum(alive.sum(axis=1), 1)
|
|
keep = coverage >= MIN_ROW_COVERAGE
|
|
if not keep.all():
|
|
dropped = [str(d.date()) for d in df.index[~keep]]
|
|
print(f"--- Dropped {len(dropped)} sparse non-trading rows: {dropped[:5]} ---")
|
|
return df[keep]
|
|
|
|
|
|
def _adjustment_ratio(old: pd.Series, new: pd.Series) -> float | None:
|
|
"""Factor to rescale older history when Yahoo re-adjusted the overlap window.
|
|
|
|
Uses the median ratio over the earliest overlapping bars so a couple of
|
|
bad stored values cannot fake an adjustment.
|
|
"""
|
|
both = pd.concat([old, new], axis=1, keys=["old", "new"]).dropna()
|
|
both = both[both["old"] > 0].head(5)
|
|
if len(both) < 2:
|
|
return None
|
|
ratio = float((both["new"] / both["old"]).median())
|
|
if not np.isfinite(ratio) or abs(ratio - 1.0) <= RESCALE_TOLERANCE:
|
|
return None
|
|
return ratio
|
|
|
|
|
|
def _merge_refresh(existing: pd.DataFrame | None, new: pd.DataFrame,
|
|
window_start: pd.Timestamp | None) -> pd.DataFrame:
|
|
"""Overwrite requested tickers from ``window_start`` on with freshly downloaded bars."""
|
|
new = new.dropna(axis=1, how="all")
|
|
if existing is None or existing.empty:
|
|
return _drop_sparse_rows(new.sort_index())
|
|
out = existing.copy()
|
|
if window_start is None:
|
|
window_start = new.index.min() if not new.empty else out.index.max()
|
|
rescaled = []
|
|
for col in new.columns:
|
|
if col not in out.columns:
|
|
continue
|
|
ratio = _adjustment_ratio(out.loc[out.index >= window_start, col], new[col])
|
|
if ratio is not None:
|
|
out.loc[out.index < window_start, col] *= ratio
|
|
rescaled.append(f"{col}x{ratio:.4f}")
|
|
if rescaled:
|
|
print(f"--- Rescaled history for re-adjusted tickers: {rescaled[:8]} ---")
|
|
out = out.reindex(index=out.index.union(new.index),
|
|
columns=list(out.columns) + [c for c in new.columns if c not in out.columns])
|
|
window = out.index >= window_start
|
|
out.loc[window, new.columns] = new.reindex(out.index[window]).to_numpy()
|
|
return _drop_sparse_rows(out.sort_index())
|
|
|
|
|
|
def _fill_open(close: pd.DataFrame, open_raw: pd.DataFrame | None) -> pd.DataFrame:
|
|
"""Align open prices to the close panel; fall back to close where open is missing."""
|
|
if open_raw is None or open_raw.empty:
|
|
return close.copy()
|
|
aligned = open_raw.reindex(index=close.index, columns=close.columns).ffill()
|
|
return aligned.fillna(close)
|
|
|
|
|
|
def update(market: str, tickers: list[str],
|
|
with_open: bool = False) -> pd.DataFrame | tuple[pd.DataFrame, pd.DataFrame]:
|
|
"""
|
|
Refresh the stored panel for ``tickers`` and return it forward-filled.
|
|
|
|
- First run: downloads 10 years of history.
|
|
- Every run: re-downloads the last ``REFRESH_DAYS`` calendar days for the
|
|
requested tickers already in the panel and overwrites them.
|
|
- New tickers are backfilled over the panel's full date range; tickers
|
|
Yahoo has no data for are remembered and retried weekly.
|
|
|
|
Returns the close panel, or ``(close, open)`` when ``with_open`` is True.
|
|
"""
|
|
fields = ["Close", "Open"] if with_open else ["Close"]
|
|
with _panel_lock(market):
|
|
close_raw = load_raw(market, "close")
|
|
open_raw = load_raw(market, "open") if with_open else None
|
|
unavailable = _load_unavailable(market)
|
|
today = datetime.now()
|
|
|
|
if close_raw is None:
|
|
start = (today - timedelta(days=365 * 10)).strftime("%Y-%m-%d")
|
|
downloaded = _download(tickers, start=start, fields=fields)
|
|
close_raw = _merge_refresh(None, downloaded["Close"], None)
|
|
if with_open:
|
|
open_raw = _merge_refresh(None, downloaded.get("Open", pd.DataFrame()), None)
|
|
else:
|
|
if with_open and open_raw is None:
|
|
print("--- Downloading Open prices for the full panel (first time) ---")
|
|
start = close_raw.index[0].strftime("%Y-%m-%d")
|
|
opens = _download(list(close_raw.columns), start=start, fields=["Open"])
|
|
open_raw = _merge_refresh(None, opens["Open"], None)
|
|
|
|
known = [t for t in tickers if t in close_raw.columns]
|
|
if known:
|
|
window_start = close_raw.index[-1] - pd.Timedelta(days=REFRESH_DAYS)
|
|
fresh = _download(known, start=window_start.strftime("%Y-%m-%d"), fields=fields)
|
|
if not fresh["Close"].empty:
|
|
close_raw = _merge_refresh(close_raw, fresh["Close"], window_start)
|
|
if with_open:
|
|
open_raw = _merge_refresh(open_raw, fresh.get("Open", pd.DataFrame()), window_start)
|
|
else:
|
|
print("--- Refresh download returned no data; keeping stored panel ---")
|
|
|
|
cutoff = (today - timedelta(days=UNAVAILABLE_RETRY_DAYS)).strftime("%Y-%m-%d")
|
|
new_tickers = [t for t in tickers
|
|
if t not in close_raw.columns and unavailable.get(t, "") < cutoff]
|
|
if new_tickers:
|
|
start = close_raw.index[0].strftime("%Y-%m-%d")
|
|
backfill = _download(new_tickers, start=start, fields=fields)
|
|
downloaded_any = not backfill["Close"].dropna(how="all").empty
|
|
got = backfill["Close"].dropna(axis=1, how="all")
|
|
got = got.reindex(index=close_raw.index)
|
|
got = got.dropna(axis=1, how="all")
|
|
for t in new_tickers:
|
|
if t in got.columns:
|
|
unavailable.pop(t, None)
|
|
elif downloaded_any:
|
|
# Only a ticker missing from an otherwise successful
|
|
# download is unavailable; a failed request is retried.
|
|
unavailable[t] = today.strftime("%Y-%m-%d")
|
|
if not got.empty:
|
|
close_raw = pd.concat([close_raw, got], axis=1)
|
|
if with_open and open_raw is not None:
|
|
opens = backfill.get("Open", pd.DataFrame())
|
|
opens = opens.reindex(index=open_raw.index, columns=got.columns)
|
|
open_raw = pd.concat([open_raw, opens], axis=1)
|
|
print(f"--- Backfilled {got.shape[1]} new tickers ---")
|
|
missing = len(new_tickers) - got.shape[1]
|
|
if missing:
|
|
print(f"--- {missing} tickers have no Yahoo data; retry after "
|
|
f"{UNAVAILABLE_RETRY_DAYS} days ---")
|
|
|
|
close_raw = close_raw.loc[:, ~close_raw.columns.duplicated()]
|
|
_write_atomic(close_raw, _data_path(market, "close"))
|
|
print(f"--- Saved {close_raw.shape[0]} days x {close_raw.shape[1]} tickers to "
|
|
f"{_data_path(market, 'close')} ---")
|
|
if with_open and open_raw is not None:
|
|
open_raw = open_raw.loc[:, ~open_raw.columns.duplicated()]
|
|
_write_atomic(open_raw, _data_path(market, "open"))
|
|
_save_unavailable(market, unavailable)
|
|
|
|
close = close_raw.ffill()
|
|
if with_open:
|
|
return close, _fill_open(close, open_raw)
|
|
return close
|
|
|
|
|
|
def rebuild(market: str, with_open: bool = True, years: int = 10) -> None:
|
|
"""Re-download the full history of every ticker in a stored panel.
|
|
|
|
Used to repair panels written by the old append-only updater. Tickers
|
|
Yahoo no longer serves keep their stored history.
|
|
"""
|
|
with _panel_lock(market):
|
|
old_close = load_raw(market, "close")
|
|
old_open = load_raw(market, "open")
|
|
if old_close is None:
|
|
raise FileNotFoundError(_data_path(market, "close"))
|
|
tickers = list(old_close.columns)
|
|
start = (datetime.now() - timedelta(days=365 * years)).strftime("%Y-%m-%d")
|
|
fields = ["Close", "Open"] if with_open else ["Close"]
|
|
fresh = _download(tickers, start=start, fields=fields)
|
|
close = _merge_refresh(None, fresh["Close"], None)
|
|
kept = [t for t in tickers if t not in close.columns]
|
|
if kept:
|
|
close = close.combine_first(old_close[kept].reindex(close.index))
|
|
print(f"--- Kept stored history for {len(kept)} tickers Yahoo no longer serves ---")
|
|
close = close[[t for t in tickers if t in close.columns]]
|
|
_write_atomic(close, _data_path(market, "close"))
|
|
print(f"--- Rebuilt {market}: {close.shape[0]} days x {close.shape[1]} tickers ---")
|
|
if with_open:
|
|
opens = fresh.get("Open", pd.DataFrame()).dropna(axis=1, how="all")
|
|
opens = opens.reindex(index=close.index)
|
|
if old_open is not None:
|
|
missing = [t for t in close.columns if t not in opens.columns and t in old_open.columns]
|
|
if missing:
|
|
opens = pd.concat([opens, old_open[missing].reindex(close.index)], axis=1)
|
|
_write_atomic(opens, _data_path(market, "open"))
|