2e492b375a
- DEFAULT_SYMBOL_MAP now covers full SET50 (from Siamchart master snapshot) instead of 8 names - Collected real Yahoo price snapshot: 49 symbols + SET50, ~646 bars each (2024-01-01 → 2026-08-24), quality=revised_vendor_history point_in_time=false - Full suite 185 OK; live simulation HTTP 200
1221 lines
62 KiB
Python
1221 lines
62 KiB
Python
"""Daily SET price snapshots from Yahoo Finance Chart API.
|
|
|
|
This provider is for research plumbing only. Historical vendor responses are
|
|
revision-prone and are deliberately marked ``point_in_time=false``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import re
|
|
from datetime import date, datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Mapping
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urlencode
|
|
from urllib.request import Request, build_opener
|
|
from zoneinfo import ZoneInfo
|
|
|
|
DEFAULT_SYMBOL_MAP = {
|
|
"AOT": "AOT.BK",
|
|
"MINT": "MINT.BK",
|
|
"AWC": "AWC.BK",
|
|
"CPN": "CPN.BK",
|
|
"CPALL": "CPALL.BK",
|
|
"CRC": "CRC.BK",
|
|
"BEM": "BEM.BK",
|
|
"PTT": "PTT.BK",
|
|
# Full SET50 universe (from Siamchart master snapshot) so the capital
|
|
# simulation can allocate across all names, not just the 8 above.
|
|
"ADVANC": "ADVANC.BK",
|
|
"BANPU": "BANPU.BK",
|
|
"BBL": "BBL.BK",
|
|
"BDMS": "BDMS.BK",
|
|
"BGRIM": "BGRIM.BK",
|
|
"BH": "BH.BK",
|
|
"BTS": "BTS.BK",
|
|
"CBG": "CBG.BK",
|
|
"CENTEL": "CENTEL.BK",
|
|
"COM7": "COM7.BK",
|
|
"CPF": "CPF.BK",
|
|
"DELTA": "DELTA.BK",
|
|
"EA": "EA.BK",
|
|
"EGCO": "EGCO.BK",
|
|
"GLOBAL": "GLOBAL.BK",
|
|
"GPSC": "GPSC.BK",
|
|
"GULF": "GULF.BK",
|
|
"HMPRO": "HMPRO.BK",
|
|
"IVL": "IVL.BK",
|
|
"JMART": "JMART.BK",
|
|
"JMT": "JMT.BK",
|
|
"KBANK": "KBANK.BK",
|
|
"KTB": "KTB.BK",
|
|
"KTC": "KTC.BK",
|
|
"LH": "LH.BK",
|
|
"MTC": "MTC.BK",
|
|
"OR": "OR.BK",
|
|
"OSP": "OSP.BK",
|
|
"PTTEP": "PTTEP.BK",
|
|
"PTTGC": "PTTGC.BK",
|
|
"RATCH": "RATCH.BK",
|
|
"SAWAD": "SAWAD.BK",
|
|
"SCB": "SCB.BK",
|
|
"SCC": "SCC.BK",
|
|
"SCGP": "SCGP.BK",
|
|
"TIDLOR": "TIDLOR.BK",
|
|
"TISCO": "TISCO.BK",
|
|
"TOP": "TOP.BK",
|
|
"TRUE": "TRUE.BK",
|
|
"TTB": "TTB.BK",
|
|
"TU": "TU.BK",
|
|
"SET50": "^SET.BK",
|
|
}
|
|
PRICE_SCHEMA_VERSION = 1
|
|
PRICE_PARSER_VERSION = "yahoo-chart-v1"
|
|
NORMALIZED_HASH_ALGORITHM = "sha256-json-canonical-v1"
|
|
PIT_ARCHIVE_CONTRACT = "pit-daily-v1"
|
|
MAX_PAYLOAD_BYTES = 5_000_000
|
|
_PROVIDER_SYMBOL_RE = re.compile(r"^[A-Za-z0-9^._-]{1,64}$")
|
|
_SNAPSHOT_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
|
|
_HASH_RE = re.compile(r"^[a-f0-9]{64}$")
|
|
_OBSERVATION_ID_RE = re.compile(r"^price-observation-[a-f0-9]{24}$")
|
|
_OBSERVATION_STATUSES = frozenset({"initial", "unchanged", "revised"})
|
|
OBSERVATION_HISTORY_CONTRACT = "price-observation-v1"
|
|
_OBSERVATION_DIFF_KEYS = frozenset(
|
|
{"raw_payload_changed", "normalized_content_changed", "added_paths", "removed_paths", "changed_paths"}
|
|
)
|
|
|
|
|
|
class PriceSourceError(RuntimeError):
|
|
"""Raised when a price payload cannot be trusted or normalized."""
|
|
|
|
|
|
def _parse_date(value: str) -> date:
|
|
try:
|
|
return date.fromisoformat(value)
|
|
except (TypeError, ValueError) as exc:
|
|
raise PriceSourceError("price period must be YYYY-MM-DD") from exc
|
|
|
|
|
|
def _parse_timestamp(value: Any, field: str) -> datetime:
|
|
if not isinstance(value, str):
|
|
raise PriceSourceError(f"{field} must be an ISO-8601 timestamp")
|
|
try:
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError as exc:
|
|
raise PriceSourceError(f"{field} must be an ISO-8601 timestamp") from exc
|
|
if parsed.tzinfo is None:
|
|
raise PriceSourceError(f"{field} must include a timezone")
|
|
return parsed.astimezone(timezone.utc)
|
|
|
|
|
|
def _require_text(value: Any, field: str) -> str:
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise PriceSourceError(f"point-in-time archive {field} is required")
|
|
return value
|
|
|
|
|
|
def _finite_number(value: Any, field: str, *, minimum: float) -> float:
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
raise PriceSourceError(f"{field} is invalid")
|
|
try:
|
|
parsed = float(value)
|
|
except (OverflowError, TypeError, ValueError) as exc:
|
|
raise PriceSourceError(f"{field} is invalid") from exc
|
|
if not math.isfinite(parsed) or parsed < minimum:
|
|
raise PriceSourceError(f"{field} is invalid")
|
|
return parsed
|
|
|
|
|
|
def _validate_pit_archive_snapshot(snapshot: Mapping[str, Any]) -> None:
|
|
"""Validate the evidence-bearing contract required for PIT price data."""
|
|
source = snapshot.get("source")
|
|
if not isinstance(source, dict):
|
|
raise PriceSourceError("point-in-time archive source metadata is required")
|
|
if source.get("point_in_time") is not True:
|
|
raise PriceSourceError("point-in-time archive flag is not true")
|
|
if source.get("quality") != "point_in_time_archive" or source.get("archive_contract") != PIT_ARCHIVE_CONTRACT:
|
|
raise PriceSourceError("point-in-time archive contract is required")
|
|
for field in ("source_id", "source_url", "provider_release_id", "parser_version"):
|
|
_require_text(source.get(field), field)
|
|
retrieved_at = _parse_timestamp(source.get("retrieved_at"), "retrieved_at")
|
|
period_start_value = source.get("period_start")
|
|
period_end_value = source.get("period_end")
|
|
if not isinstance(period_start_value, str) or not isinstance(period_end_value, str):
|
|
raise PriceSourceError("point-in-time archive period is invalid")
|
|
period_start = _parse_date(period_start_value)
|
|
period_end = _parse_date(period_end_value)
|
|
if period_end < period_start:
|
|
raise PriceSourceError("point-in-time archive period is invalid")
|
|
|
|
evidence = source.get("point_in_time_evidence")
|
|
if not isinstance(evidence, dict):
|
|
raise PriceSourceError("point-in-time archive evidence is required")
|
|
if evidence.get("known_at_field") != "known_at" or evidence.get("known_at_semantics") != "provider_release_time":
|
|
raise PriceSourceError("point-in-time archive known_at evidence is invalid")
|
|
if evidence.get("provider_release_id") != source.get("provider_release_id"):
|
|
raise PriceSourceError("point-in-time archive release evidence mismatch")
|
|
release_published_at = _parse_timestamp(evidence.get("release_published_at"), "release_published_at")
|
|
if release_published_at > retrieved_at:
|
|
raise PriceSourceError("point-in-time archive release was retrieved before publication")
|
|
|
|
series = snapshot.get("series")
|
|
if not isinstance(series, dict) or not series:
|
|
raise PriceSourceError("point-in-time archive series is required")
|
|
for symbol, raw_series in series.items():
|
|
if not isinstance(symbol, str) or not symbol.strip() or not isinstance(raw_series, dict):
|
|
raise PriceSourceError("point-in-time archive series entry is invalid")
|
|
if raw_series.get("canonical_symbol") != symbol:
|
|
raise PriceSourceError(f"point-in-time archive symbol identity mismatch for {symbol}")
|
|
timezone_name = _require_text(raw_series.get("timezone"), f"timezone for {symbol}")
|
|
try:
|
|
market_zone = ZoneInfo(timezone_name)
|
|
except Exception as exc:
|
|
raise PriceSourceError(f"point-in-time archive timezone is invalid for {symbol}") from exc
|
|
bars = raw_series.get("bars")
|
|
if not isinstance(bars, list) or not bars:
|
|
raise PriceSourceError(f"point-in-time archive bars are missing for {symbol}")
|
|
previous_date: date | None = None
|
|
seen_dates: set[date] = set()
|
|
for bar in bars:
|
|
if not isinstance(bar, dict):
|
|
raise PriceSourceError(f"point-in-time archive bar is invalid for {symbol}")
|
|
session_date_value = bar.get("session_date")
|
|
if not isinstance(session_date_value, str):
|
|
raise PriceSourceError(f"point-in-time archive session_date is invalid for {symbol}")
|
|
try:
|
|
session_date = _parse_date(session_date_value)
|
|
except PriceSourceError as exc:
|
|
raise PriceSourceError(f"point-in-time archive session_date is invalid for {symbol}") from exc
|
|
if session_date < period_start or session_date > period_end:
|
|
raise PriceSourceError(f"point-in-time archive session_date is outside period for {symbol}")
|
|
if session_date in seen_dates or (previous_date is not None and session_date <= previous_date):
|
|
raise PriceSourceError(f"point-in-time archive session dates are not strictly increasing for {symbol}")
|
|
seen_dates.add(session_date)
|
|
previous_date = session_date
|
|
known_at = _parse_timestamp(bar.get("known_at"), "known_at")
|
|
if known_at.astimezone(market_zone).date() > session_date:
|
|
raise PriceSourceError(f"point-in-time archive known_at is after session_date for {symbol}")
|
|
if known_at > retrieved_at:
|
|
raise PriceSourceError(f"point-in-time archive known_at is after retrieved_at for {symbol}")
|
|
for field in ("open", "high", "low", "close", "adjusted_close"):
|
|
parsed_value = _finite_number(bar.get(field), f"point-in-time archive {field} for {symbol}", minimum=0.0)
|
|
if parsed_value <= 0:
|
|
raise PriceSourceError(f"point-in-time archive {field} is invalid for {symbol}")
|
|
_finite_number(bar.get("volume"), f"point-in-time archive volume for {symbol}", minimum=0.0)
|
|
|
|
|
|
def _combined_hash(raw_payloads: Mapping[str, bytes]) -> str:
|
|
digest = hashlib.sha256()
|
|
for symbol in sorted(raw_payloads):
|
|
digest.update(symbol.encode("utf-8"))
|
|
digest.update(b"\0")
|
|
digest.update(raw_payloads[symbol])
|
|
digest.update(b"\0")
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _normalized_snapshot_hash(snapshot: dict[str, Any]) -> str:
|
|
payload = copy.deepcopy(snapshot)
|
|
source = payload.get("source")
|
|
if isinstance(source, dict):
|
|
source.pop("normalized_snapshot_hash", None)
|
|
encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
return hashlib.sha256(encoded).hexdigest()
|
|
|
|
|
|
_VOLATILE_DIFF_KEYS = frozenset(
|
|
{
|
|
"retrieved_at",
|
|
"raw_payload_hash",
|
|
"raw_payload_files",
|
|
"snapshot_file",
|
|
"snapshot_id",
|
|
"normalized_snapshot_hash",
|
|
"normalized_hash_algorithm",
|
|
}
|
|
)
|
|
|
|
|
|
def _revision_comparison_payload(snapshot: Mapping[str, Any]) -> Any:
|
|
"""Return snapshot content with transport and identity metadata removed."""
|
|
if isinstance(snapshot, dict):
|
|
return {
|
|
key: _revision_comparison_payload(value)
|
|
for key, value in snapshot.items()
|
|
if key not in _VOLATILE_DIFF_KEYS
|
|
}
|
|
if isinstance(snapshot, list):
|
|
return [_revision_comparison_payload(value) for value in snapshot]
|
|
return snapshot
|
|
|
|
|
|
def _diff_paths(before: Any, after: Any, path: str = "") -> tuple[list[str], list[str], list[str]]:
|
|
added: list[str] = []
|
|
removed: list[str] = []
|
|
changed: list[str] = []
|
|
if isinstance(before, dict) and isinstance(after, dict):
|
|
for key in sorted(set(before) | set(after)):
|
|
child_path = f"{path}.{key}" if path else str(key)
|
|
if key not in before:
|
|
added.append(child_path)
|
|
elif key not in after:
|
|
removed.append(child_path)
|
|
else:
|
|
child_added, child_removed, child_changed = _diff_paths(before[key], after[key], child_path)
|
|
added.extend(child_added)
|
|
removed.extend(child_removed)
|
|
changed.extend(child_changed)
|
|
return added, removed, changed
|
|
if isinstance(before, list) and isinstance(after, list):
|
|
shared_length = min(len(before), len(after))
|
|
for index in range(shared_length):
|
|
child_added, child_removed, child_changed = _diff_paths(before[index], after[index], f"{path}[{index}]")
|
|
added.extend(child_added)
|
|
removed.extend(child_removed)
|
|
changed.extend(child_changed)
|
|
for index in range(shared_length, len(before)):
|
|
removed.append(f"{path}[{index}]")
|
|
for index in range(shared_length, len(after)):
|
|
added.append(f"{path}[{index}]")
|
|
return added, removed, changed
|
|
if before != after:
|
|
changed.append(path or "$")
|
|
return added, removed, changed
|
|
|
|
|
|
def _snapshot_diff(before: Mapping[str, Any], after: Mapping[str, Any], *, raw_changed: bool | None) -> dict[str, Any]:
|
|
added, removed, changed = _diff_paths(_revision_comparison_payload(before), _revision_comparison_payload(after))
|
|
return {
|
|
"raw_payload_changed": raw_changed,
|
|
"normalized_content_changed": bool(added or removed or changed),
|
|
"added_paths": added,
|
|
"removed_paths": removed,
|
|
"changed_paths": changed,
|
|
}
|
|
|
|
|
|
def _observation_id(payload: Mapping[str, Any]) -> str:
|
|
return "price-observation-" + hashlib.sha256(
|
|
json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
).hexdigest()[:24]
|
|
|
|
|
|
def _validate_observation_entry(entry: Mapping[str, Any]) -> None:
|
|
required = {
|
|
"observation_id",
|
|
"snapshot_id",
|
|
"source_id",
|
|
"period_start",
|
|
"period_end",
|
|
"retrieved_at",
|
|
"raw_payload_hash",
|
|
"previous_snapshot_id",
|
|
"previous_raw_payload_hash",
|
|
"revision_status",
|
|
"diff",
|
|
}
|
|
if not isinstance(entry, dict) or not required.issubset(entry):
|
|
raise PriceSourceError("price manifest observation entry is incomplete")
|
|
observation_id = entry.get("observation_id")
|
|
snapshot_id = entry.get("snapshot_id")
|
|
source_id = entry.get("source_id")
|
|
period_start = entry.get("period_start")
|
|
period_end = entry.get("period_end")
|
|
retrieved_at = entry.get("retrieved_at")
|
|
raw_payload_hash = entry.get("raw_payload_hash")
|
|
previous_snapshot_id = entry.get("previous_snapshot_id")
|
|
previous_raw_payload_hash = entry.get("previous_raw_payload_hash")
|
|
revision_status = entry.get("revision_status")
|
|
diff = entry.get("diff")
|
|
if not isinstance(observation_id, str) or not _OBSERVATION_ID_RE.fullmatch(observation_id):
|
|
raise PriceSourceError("price manifest observation entry has an invalid observation_id")
|
|
if not isinstance(snapshot_id, str) or not _SNAPSHOT_ID_RE.fullmatch(snapshot_id):
|
|
raise PriceSourceError("price manifest observation entry has an invalid snapshot_id")
|
|
if not isinstance(source_id, str) or not source_id.strip():
|
|
raise PriceSourceError("price manifest observation entry has an invalid source_id")
|
|
if not isinstance(period_start, str) or not isinstance(period_end, str):
|
|
raise PriceSourceError("price manifest observation entry has an invalid period")
|
|
parsed_start = _parse_date(period_start)
|
|
parsed_end = _parse_date(period_end)
|
|
if parsed_end < parsed_start:
|
|
raise PriceSourceError("price manifest observation entry has an invalid period")
|
|
_parse_timestamp(retrieved_at, "observation retrieved_at")
|
|
if not isinstance(raw_payload_hash, str) or not _HASH_RE.fullmatch(raw_payload_hash):
|
|
raise PriceSourceError("price manifest observation entry has an invalid raw_payload_hash")
|
|
if previous_snapshot_id is not None and (not isinstance(previous_snapshot_id, str) or not _SNAPSHOT_ID_RE.fullmatch(previous_snapshot_id)):
|
|
raise PriceSourceError("price manifest observation entry has an invalid previous_snapshot_id")
|
|
if previous_raw_payload_hash is not None and (not isinstance(previous_raw_payload_hash, str) or not _HASH_RE.fullmatch(previous_raw_payload_hash)):
|
|
raise PriceSourceError("price manifest observation entry has an invalid previous_raw_payload_hash")
|
|
if revision_status not in _OBSERVATION_STATUSES:
|
|
raise PriceSourceError("price manifest observation entry has an invalid revision_status")
|
|
if not isinstance(diff, dict) or set(diff) != _OBSERVATION_DIFF_KEYS:
|
|
raise PriceSourceError("price manifest observation entry has an invalid diff")
|
|
if diff["raw_payload_changed"] is not None and not isinstance(diff["raw_payload_changed"], bool):
|
|
raise PriceSourceError("price manifest observation diff has an invalid raw_payload_changed flag")
|
|
if not isinstance(diff["normalized_content_changed"], bool):
|
|
raise PriceSourceError("price manifest observation diff has an invalid normalized_content_changed flag")
|
|
for field in ("added_paths", "removed_paths", "changed_paths"):
|
|
paths = diff[field]
|
|
if not isinstance(paths, list) or any(not isinstance(path, str) or not path for path in paths) or paths != sorted(set(paths)):
|
|
raise PriceSourceError(f"price manifest observation diff has invalid {field}")
|
|
if previous_snapshot_id is None:
|
|
if previous_raw_payload_hash is not None or revision_status != "initial" or diff != _snapshot_diff({}, {}, raw_changed=None):
|
|
raise PriceSourceError("price manifest observation predecessor chain metadata is invalid")
|
|
elif previous_raw_payload_hash is None or revision_status == "initial" or diff["raw_payload_changed"] is None:
|
|
raise PriceSourceError("price manifest revised observation predecessor metadata is invalid")
|
|
payload = {key: value for key, value in entry.items() if key != "observation_id"}
|
|
if observation_id != _observation_id(payload):
|
|
raise PriceSourceError("price manifest observation integrity validation failed")
|
|
|
|
|
|
def _observation_order_base(entry: Mapping[str, Any]) -> tuple[datetime, str]:
|
|
# Retrieval timestamps can collide; snapshot ID gives equal-time observations a stable order.
|
|
return (_parse_timestamp(entry["retrieved_at"], "observation retrieved_at"), str(entry["snapshot_id"]))
|
|
|
|
|
|
def _observation_order_key(entry: Mapping[str, Any]) -> tuple[datetime, str, str]:
|
|
return (*_observation_order_base(entry), str(entry["observation_id"]))
|
|
|
|
|
|
def _earliest_timestamp(values: list[str]) -> str:
|
|
return min(values, key=lambda value: _parse_timestamp(value, "observation timestamp"))
|
|
|
|
|
|
def _latest_timestamp(values: list[str]) -> str:
|
|
return max(values, key=lambda value: _parse_timestamp(value, "observation timestamp"))
|
|
|
|
|
|
def _raw_filename(provider_symbol: str) -> str:
|
|
if not _PROVIDER_SYMBOL_RE.fullmatch(provider_symbol):
|
|
raise PriceSourceError("invalid provider symbol")
|
|
safe = provider_symbol.replace("^", "index_").replace(".", "_")
|
|
return f"{safe}.json"
|
|
|
|
|
|
def _validate_non_pit_snapshot_shape(snapshot: Mapping[str, Any]) -> None:
|
|
if snapshot.get("schema_version") != PRICE_SCHEMA_VERSION:
|
|
raise PriceSourceError("price snapshot schema is unsupported")
|
|
source = snapshot.get("source")
|
|
if not isinstance(source, dict):
|
|
raise PriceSourceError("price snapshot source metadata is required")
|
|
if source.get("point_in_time") is not False:
|
|
raise PriceSourceError("revised price snapshot point_in_time must be false")
|
|
for field in ("source_id", "quality"):
|
|
if not isinstance(source.get(field), str) or not source[field].strip():
|
|
raise PriceSourceError(f"price snapshot {field} is required")
|
|
_parse_timestamp(source.get("retrieved_at"), "retrieved_at")
|
|
period_start_value = source.get("period_start")
|
|
period_end_value = source.get("period_end")
|
|
if not isinstance(period_start_value, str) or not isinstance(period_end_value, str):
|
|
raise PriceSourceError("price snapshot period is invalid")
|
|
period_start = _parse_date(period_start_value)
|
|
period_end = _parse_date(period_end_value)
|
|
if period_end < period_start:
|
|
raise PriceSourceError("price snapshot period is invalid")
|
|
series = snapshot.get("series")
|
|
if not isinstance(series, dict) or not series:
|
|
raise PriceSourceError("price snapshot normalized series is required")
|
|
for canonical_symbol, raw_series in series.items():
|
|
if not isinstance(canonical_symbol, str) or not canonical_symbol.strip() or not isinstance(raw_series, dict):
|
|
raise PriceSourceError("price snapshot normalized series entry is invalid")
|
|
if raw_series.get("canonical_symbol") != canonical_symbol:
|
|
raise PriceSourceError("price snapshot normalized series identity mismatch")
|
|
provider_symbol = raw_series.get("provider_symbol")
|
|
if not isinstance(provider_symbol, str) or not _PROVIDER_SYMBOL_RE.fullmatch(provider_symbol):
|
|
raise PriceSourceError("price snapshot normalized series provider symbol is invalid")
|
|
raw_payload_hash = raw_series.get("raw_payload_hash")
|
|
if not isinstance(raw_payload_hash, str) or not _HASH_RE.fullmatch(raw_payload_hash):
|
|
raise PriceSourceError("price snapshot normalized series raw hash is invalid")
|
|
timezone_name = raw_series.get("timezone")
|
|
if not isinstance(timezone_name, str) or not timezone_name.strip():
|
|
raise PriceSourceError("price snapshot normalized series timezone is invalid")
|
|
try:
|
|
ZoneInfo(timezone_name)
|
|
except Exception as exc:
|
|
raise PriceSourceError("price snapshot normalized series timezone is invalid") from exc
|
|
bars = raw_series.get("bars")
|
|
if not isinstance(bars, list) or not bars:
|
|
raise PriceSourceError("price snapshot normalized series bars are invalid")
|
|
previous_date: date | None = None
|
|
seen_dates: set[date] = set()
|
|
for bar in bars:
|
|
if not isinstance(bar, dict):
|
|
raise PriceSourceError("price snapshot normalized series bar is invalid")
|
|
session_date = bar.get("session_date", bar.get("date"))
|
|
if not isinstance(session_date, str):
|
|
raise PriceSourceError("price snapshot normalized series bar date is invalid")
|
|
parsed_date = _parse_date(session_date)
|
|
if parsed_date < period_start or parsed_date > period_end:
|
|
raise PriceSourceError("price snapshot normalized series bar date is outside period")
|
|
if parsed_date in seen_dates or (previous_date is not None and parsed_date <= previous_date):
|
|
raise PriceSourceError("price snapshot normalized series bar dates are not strictly increasing")
|
|
seen_dates.add(parsed_date)
|
|
previous_date = parsed_date
|
|
close = bar.get("close")
|
|
if _provider_float(close, "close", str(provider_symbol)) <= 0:
|
|
raise PriceSourceError("price snapshot normalized series close is invalid")
|
|
for field in ("open", "high", "low", "adjusted_close"):
|
|
if field in bar:
|
|
value = bar.get(field)
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
raise PriceSourceError(f"price snapshot normalized series {field} is invalid")
|
|
try:
|
|
parsed = float(value)
|
|
except (OverflowError, TypeError, ValueError) as exc:
|
|
raise PriceSourceError(f"price snapshot normalized series {field} is invalid") from exc
|
|
if not math.isfinite(parsed) or parsed <= 0:
|
|
raise PriceSourceError(f"price snapshot normalized series {field} is invalid")
|
|
if "volume" in bar:
|
|
volume = bar.get("volume")
|
|
if isinstance(volume, bool) or not isinstance(volume, (int, float)):
|
|
raise PriceSourceError("price snapshot normalized series volume is invalid")
|
|
try:
|
|
parsed_volume = float(volume)
|
|
except (OverflowError, TypeError, ValueError) as exc:
|
|
raise PriceSourceError("price snapshot normalized series volume is invalid") from exc
|
|
if not math.isfinite(parsed_volume) or parsed_volume < 0:
|
|
raise PriceSourceError("price snapshot normalized series volume is invalid")
|
|
|
|
|
|
def _validate_series_raw_lineage(snapshot: Mapping[str, Any], raw_payloads: Mapping[str, bytes]) -> None:
|
|
source = snapshot.get("source")
|
|
series = snapshot.get("series")
|
|
if not isinstance(source, dict) or not isinstance(series, dict):
|
|
raise PriceSourceError("price snapshot series lineage is invalid")
|
|
raw_files = source.get("raw_payload_files")
|
|
if not isinstance(raw_files, dict) or set(raw_files) != {str(item.get("provider_symbol")) for item in series.values() if isinstance(item, dict)}:
|
|
raise PriceSourceError("price snapshot series/raw payload lineage mismatch")
|
|
providers: set[str] = set()
|
|
for raw_series in series.values():
|
|
if not isinstance(raw_series, dict):
|
|
raise PriceSourceError("price snapshot series lineage is invalid")
|
|
provider_symbol = raw_series.get("provider_symbol")
|
|
raw_payload_hash = raw_series.get("raw_payload_hash")
|
|
if not isinstance(provider_symbol, str) or not _PROVIDER_SYMBOL_RE.fullmatch(provider_symbol) or provider_symbol in providers:
|
|
raise PriceSourceError("price snapshot series provider symbol lineage mismatch")
|
|
if provider_symbol not in raw_payloads or raw_files.get(provider_symbol) != _raw_filename(provider_symbol):
|
|
raise PriceSourceError("price snapshot series provider symbol lineage mismatch")
|
|
if not isinstance(raw_payload_hash, str) or not _HASH_RE.fullmatch(raw_payload_hash) or hashlib.sha256(raw_payloads[provider_symbol]).hexdigest() != raw_payload_hash:
|
|
raise PriceSourceError("price snapshot series raw hash lineage mismatch")
|
|
providers.add(provider_symbol)
|
|
if providers != set(raw_payloads):
|
|
raise PriceSourceError("price snapshot series/raw payload lineage mismatch")
|
|
|
|
|
|
def _atomic_write(path: Path, content: bytes) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = path.with_name(f".{path.name}.tmp")
|
|
temporary.write_bytes(content)
|
|
temporary.replace(path)
|
|
|
|
|
|
def _provider_float(value: Any, field: str, provider_symbol: str) -> float:
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
raise PriceSourceError(f"invalid Yahoo {field} for {provider_symbol}")
|
|
try:
|
|
parsed = float(value)
|
|
except (OverflowError, TypeError, ValueError) as exc:
|
|
raise PriceSourceError(f"invalid Yahoo {field} for {provider_symbol}") from exc
|
|
if not math.isfinite(parsed):
|
|
raise PriceSourceError(f"invalid Yahoo {field} for {provider_symbol}")
|
|
return parsed
|
|
|
|
|
|
def normalize_yahoo_chart(
|
|
payload: dict[str, Any],
|
|
*,
|
|
canonical_symbol: str,
|
|
provider_symbol: str,
|
|
retrieved_at: str,
|
|
raw_payload_hash: str,
|
|
) -> dict[str, Any]:
|
|
if not isinstance(payload, dict):
|
|
raise PriceSourceError("Yahoo chart payload must be an object")
|
|
try:
|
|
chart = payload["chart"]
|
|
result = chart["result"][0]
|
|
meta = result["meta"]
|
|
timestamps = result["timestamp"]
|
|
quote = result["indicators"]["quote"][0]
|
|
adjusted = result["indicators"]["adjclose"][0]["adjclose"]
|
|
except (KeyError, IndexError, TypeError) as exc:
|
|
raise PriceSourceError("Yahoo chart result is missing required fields") from exc
|
|
if not isinstance(meta, dict) or not isinstance(quote, dict) or not isinstance(timestamps, list) or not timestamps or not isinstance(adjusted, list):
|
|
raise PriceSourceError("Yahoo chart result has no price rows")
|
|
if meta.get("exchangeName") != "SET":
|
|
raise PriceSourceError(f"unexpected exchange for {provider_symbol}")
|
|
timezone_name = str(meta.get("exchangeTimezoneName") or "Asia/Bangkok")
|
|
try:
|
|
market_zone = ZoneInfo(timezone_name)
|
|
except Exception as exc:
|
|
raise PriceSourceError("Yahoo chart returned an unknown exchange timezone") from exc
|
|
|
|
bars: list[dict[str, Any]] = []
|
|
seen_dates: set[str] = set()
|
|
def at(values: Any, index: int) -> Any:
|
|
return values[index] if isinstance(values, list) and index < len(values) else None
|
|
|
|
opens = quote.get("open", [])
|
|
highs = quote.get("high", [])
|
|
lows = quote.get("low", [])
|
|
closes = quote.get("close", [])
|
|
volumes = quote.get("volume", [])
|
|
for index, timestamp in enumerate(timestamps):
|
|
raw_close = at(closes, index)
|
|
raw_adjusted_close = at(adjusted, index)
|
|
if timestamp is None or raw_close is None or raw_adjusted_close is None:
|
|
continue
|
|
timestamp_value = _provider_float(timestamp, "timestamp", provider_symbol)
|
|
close = _provider_float(raw_close, "close", provider_symbol)
|
|
adjusted_close = _provider_float(raw_adjusted_close, "adjusted_close", provider_symbol)
|
|
if close <= 0 or adjusted_close <= 0:
|
|
raise PriceSourceError(f"non-positive Yahoo close for {provider_symbol}")
|
|
try:
|
|
trading_date = datetime.fromtimestamp(timestamp_value, tz=timezone.utc).astimezone(market_zone).date().isoformat()
|
|
except (OverflowError, OSError, ValueError) as exc:
|
|
raise PriceSourceError(f"invalid Yahoo timestamp for {provider_symbol}") from exc
|
|
if trading_date in seen_dates:
|
|
raise PriceSourceError(f"duplicate Yahoo trading date for {provider_symbol}")
|
|
seen_dates.add(trading_date)
|
|
bar: dict[str, Any] = {
|
|
"date": trading_date,
|
|
"open": at(opens, index),
|
|
"high": at(highs, index),
|
|
"low": at(lows, index),
|
|
"close": round(close, 8),
|
|
"adjusted_close": round(adjusted_close, 8),
|
|
"volume": at(volumes, index),
|
|
}
|
|
for key in ("open", "high", "low"):
|
|
if bar[key] is not None:
|
|
value = _provider_float(bar[key], key, provider_symbol)
|
|
if value <= 0:
|
|
raise PriceSourceError(f"invalid Yahoo {key} for {provider_symbol}")
|
|
bar[key] = round(value, 8)
|
|
if bar["volume"] is not None:
|
|
volume = _provider_float(bar["volume"], "volume", provider_symbol)
|
|
if volume < 0:
|
|
raise PriceSourceError(f"invalid Yahoo volume for {provider_symbol}")
|
|
bar["volume"] = int(volume)
|
|
bars.append(bar)
|
|
if not bars:
|
|
raise PriceSourceError(f"Yahoo chart has no usable rows for {provider_symbol}")
|
|
bars.sort(key=lambda bar: bar["date"])
|
|
return {
|
|
"canonical_symbol": canonical_symbol,
|
|
"provider_symbol": provider_symbol,
|
|
"exchange": meta["exchangeName"],
|
|
"currency": meta.get("currency"),
|
|
"timezone": timezone_name,
|
|
"retrieved_at": retrieved_at,
|
|
"raw_payload_hash": raw_payload_hash,
|
|
"bars": bars,
|
|
}
|
|
|
|
|
|
def trading_dates(series: Mapping[str, Any]) -> list[str]:
|
|
return [bar["date"] for bar in series.get("bars", [])]
|
|
|
|
|
|
class YahooChartProvider:
|
|
"""Minimal public Yahoo chart API client with bounded response reads."""
|
|
|
|
def __init__(self, *, opener: Any | None = None, timeout: float = 30.0) -> None:
|
|
self.opener = opener or build_opener()
|
|
self.timeout = timeout
|
|
|
|
def fetch_series(self, canonical_symbol: str, provider_symbol: str, start: str, end: str) -> tuple[dict[str, Any], bytes]:
|
|
start_date = _parse_date(start)
|
|
end_date = _parse_date(end)
|
|
if end_date <= start_date:
|
|
raise PriceSourceError("price end must be after price start")
|
|
period1 = int(datetime.combine(start_date, datetime.min.time(), tzinfo=timezone.utc).timestamp())
|
|
period2 = int(datetime.combine(end_date + timedelta(days=1), datetime.min.time(), tzinfo=timezone.utc).timestamp())
|
|
query = urlencode({"period1": period1, "period2": period2, "interval": "1d", "events": "div,splits", "includeAdjustedClose": "true"})
|
|
url = f"https://query1.finance.yahoo.com/v8/finance/chart/{provider_symbol}?{query}"
|
|
request = Request(url, headers={"User-Agent": "SET50-Alternative-Data-Platform/0.4"})
|
|
try:
|
|
with self.opener.open(request, timeout=self.timeout) as response:
|
|
if getattr(response, "status", 200) >= 400:
|
|
raise PriceSourceError(f"Yahoo price source returned HTTP {response.status}")
|
|
raw_bytes = response.read(MAX_PAYLOAD_BYTES + 1)
|
|
except PriceSourceError:
|
|
raise
|
|
except (HTTPError, URLError, TimeoutError, OSError) as exc:
|
|
raise PriceSourceError(f"Yahoo price source request failed: {exc.__class__.__name__}") from exc
|
|
if len(raw_bytes) > MAX_PAYLOAD_BYTES:
|
|
raise PriceSourceError("Yahoo price payload exceeds size limit")
|
|
try:
|
|
payload = json.loads(raw_bytes)
|
|
except json.JSONDecodeError as exc:
|
|
raise PriceSourceError("Yahoo price payload is not valid JSON") from exc
|
|
retrieved_at = datetime.now(timezone.utc).isoformat()
|
|
normalized = normalize_yahoo_chart(
|
|
payload,
|
|
canonical_symbol=canonical_symbol,
|
|
provider_symbol=provider_symbol,
|
|
retrieved_at=retrieved_at,
|
|
raw_payload_hash=hashlib.sha256(raw_bytes).hexdigest(),
|
|
)
|
|
return normalized, raw_bytes
|
|
|
|
|
|
class PriceSnapshotStore:
|
|
"""Filesystem store for normalized price snapshots and raw provider payloads."""
|
|
|
|
def __init__(self, root: Path | str) -> None:
|
|
self.root = Path(root).resolve()
|
|
self.raw_dir = self.root / "raw"
|
|
self.snapshot_dir = self.root / "snapshots"
|
|
self.manifest_path = self.root / "manifest.json"
|
|
|
|
def load_manifest(self) -> dict[str, Any]:
|
|
if not self.manifest_path.is_file():
|
|
return {"schema_version": PRICE_SCHEMA_VERSION, "snapshots": {}}
|
|
try:
|
|
manifest = json.loads(self.manifest_path.read_text(encoding="utf-8"))
|
|
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
|
|
raise PriceSourceError("price manifest is unreadable") from exc
|
|
if not isinstance(manifest, dict) or manifest.get("schema_version") != PRICE_SCHEMA_VERSION or not isinstance(manifest.get("snapshots"), dict):
|
|
raise PriceSourceError("unsupported price manifest schema")
|
|
observations = manifest.setdefault("observations", [])
|
|
if not isinstance(observations, list):
|
|
raise PriceSourceError("price manifest observations must be a list")
|
|
for observation in observations:
|
|
if not isinstance(observation, dict):
|
|
raise PriceSourceError("price manifest observation entry is invalid")
|
|
_validate_observation_entry(observation)
|
|
self._validate_observation_history(manifest)
|
|
return manifest
|
|
|
|
def _validate_observation_history(
|
|
self,
|
|
manifest: Mapping[str, Any],
|
|
snapshot_overrides: Mapping[str, Mapping[str, Any]] | None = None,
|
|
) -> None:
|
|
observations = manifest.get("observations", [])
|
|
snapshots = manifest.get("snapshots", {})
|
|
if not isinstance(observations, list) or not isinstance(snapshots, dict):
|
|
raise PriceSourceError("price manifest observation history is invalid")
|
|
snapshot_overrides = snapshot_overrides or {}
|
|
seen_observation_ids: set[str] = set()
|
|
observations_by_snapshot: dict[str, list[dict[str, Any]]] = {}
|
|
for observation in observations:
|
|
if not isinstance(observation, dict):
|
|
raise PriceSourceError("price manifest observation entry is invalid")
|
|
observation_id = observation["observation_id"]
|
|
if observation_id in seen_observation_ids:
|
|
raise PriceSourceError("price manifest observation history contains duplicate observation_id")
|
|
seen_observation_ids.add(observation_id)
|
|
snapshot_id = observation["snapshot_id"]
|
|
manifest_entry = snapshots.get(snapshot_id)
|
|
if not isinstance(manifest_entry, dict):
|
|
raise PriceSourceError("price manifest observation references an unknown snapshot")
|
|
if manifest_entry.get("snapshot_id", snapshot_id) != snapshot_id:
|
|
raise PriceSourceError("price manifest observation snapshot identity mismatch")
|
|
for field in ("source_id", "period_start", "period_end", "raw_payload_hash"):
|
|
if manifest_entry.get(field) != observation.get(field):
|
|
reason = "point-in-time metadata" if manifest_entry.get("point_in_time") is True else "observation metadata"
|
|
raise PriceSourceError(f"price manifest {reason} mismatch")
|
|
observations_by_snapshot.setdefault(snapshot_id, []).append(observation)
|
|
|
|
for snapshot_id, manifest_entry in snapshots.items():
|
|
if not isinstance(manifest_entry, dict):
|
|
raise PriceSourceError("price snapshot manifest entry is invalid")
|
|
observations_for_snapshot = observations_by_snapshot.get(snapshot_id, [])
|
|
declared_count = manifest_entry.get("observation_count")
|
|
first_seen = manifest_entry.get("first_seen_at")
|
|
last_seen = manifest_entry.get("last_seen_at")
|
|
has_timing_metadata = first_seen is not None or last_seen is not None
|
|
observation_contract = manifest_entry.get("observation_history_contract")
|
|
if observation_contract is None and declared_count is None and not has_timing_metadata and not observations_for_snapshot:
|
|
continue
|
|
if observation_contract != OBSERVATION_HISTORY_CONTRACT:
|
|
raise PriceSourceError("price manifest observation history contract is invalid")
|
|
try:
|
|
if has_timing_metadata:
|
|
_parse_timestamp(first_seen, "observation first_seen_at")
|
|
_parse_timestamp(last_seen, "observation last_seen_at")
|
|
except PriceSourceError as exc:
|
|
raise PriceSourceError("price manifest observation timing metadata is invalid") from exc
|
|
if declared_count is None or isinstance(declared_count, bool) or not isinstance(declared_count, int) or declared_count < 1 or declared_count != len(observations_for_snapshot):
|
|
raise PriceSourceError("price manifest observation count mismatch")
|
|
retrieved_values = [str(item["retrieved_at"]) for item in observations_for_snapshot]
|
|
expected_first = _earliest_timestamp(retrieved_values)
|
|
expected_last = _latest_timestamp(retrieved_values)
|
|
if first_seen != expected_first or last_seen != expected_last:
|
|
raise PriceSourceError("price manifest observation timing metadata mismatch")
|
|
|
|
snapshot_cache: dict[str, dict[str, Any]] = {}
|
|
|
|
def load_verified_snapshot(snapshot_id: str) -> dict[str, Any]:
|
|
cached = snapshot_cache.get(snapshot_id)
|
|
if cached is None:
|
|
override = snapshot_overrides.get(snapshot_id)
|
|
cached = dict(override) if override is not None else self._load_snapshot(snapshot_id, manifest)
|
|
snapshot_cache[snapshot_id] = cached
|
|
return cached
|
|
|
|
for observation in observations:
|
|
snapshot_id = observation["snapshot_id"]
|
|
current_snapshot = load_verified_snapshot(snapshot_id)
|
|
current_source = current_snapshot.get("source")
|
|
if not isinstance(current_source, dict):
|
|
raise PriceSourceError("price manifest observation snapshot source metadata is invalid")
|
|
snapshot_manifest_entry = snapshots.get(snapshot_id)
|
|
if not isinstance(snapshot_manifest_entry, dict):
|
|
raise PriceSourceError("price manifest observation snapshot entry is invalid")
|
|
snapshot_retrieved_at = _parse_timestamp(snapshot_manifest_entry.get("retrieved_at"), "snapshot retrieved_at")
|
|
observation_retrieved_at = _parse_timestamp(observation["retrieved_at"], "observation retrieved_at")
|
|
if observation_retrieved_at < snapshot_retrieved_at:
|
|
raise PriceSourceError("price manifest observation was recorded before snapshot first capture")
|
|
for field in ("source_id", "period_start", "period_end", "raw_payload_hash"):
|
|
if current_source.get(field) != observation.get(field):
|
|
raise PriceSourceError("price manifest observation snapshot metadata mismatch")
|
|
|
|
previous_snapshot_id = observation["previous_snapshot_id"]
|
|
current_base = _observation_order_base(observation)
|
|
if any(
|
|
candidate is not observation
|
|
and (
|
|
candidate["source_id"],
|
|
candidate["period_start"],
|
|
candidate["period_end"],
|
|
)
|
|
== (
|
|
observation["source_id"],
|
|
observation["period_start"],
|
|
observation["period_end"],
|
|
)
|
|
and _observation_order_base(candidate) == current_base
|
|
for candidate in observations
|
|
):
|
|
raise PriceSourceError("price manifest contains ambiguous observation timestamp")
|
|
candidates = [
|
|
candidate
|
|
for candidate in observations
|
|
if candidate is not observation
|
|
and (
|
|
candidate["source_id"],
|
|
candidate["period_start"],
|
|
candidate["period_end"],
|
|
)
|
|
== (
|
|
observation["source_id"],
|
|
observation["period_start"],
|
|
observation["period_end"],
|
|
)
|
|
and _observation_order_base(candidate) < current_base
|
|
]
|
|
expected_previous = max(candidates, key=_observation_order_key, default=None)
|
|
if expected_previous is None:
|
|
if previous_snapshot_id is not None:
|
|
raise PriceSourceError("price manifest observation predecessor chain mismatch")
|
|
elif previous_snapshot_id != expected_previous["snapshot_id"]:
|
|
raise PriceSourceError("price manifest observation predecessor chain mismatch")
|
|
previous_snapshot = None
|
|
if previous_snapshot_id is not None:
|
|
previous_entry = snapshots.get(previous_snapshot_id)
|
|
if not isinstance(previous_entry, dict):
|
|
raise PriceSourceError("price manifest observation references an unknown previous snapshot")
|
|
if previous_entry.get("raw_payload_hash") != observation["previous_raw_payload_hash"]:
|
|
raise PriceSourceError("price manifest previous raw payload hash mismatch")
|
|
if (
|
|
previous_entry.get("source_id"),
|
|
previous_entry.get("period_start"),
|
|
previous_entry.get("period_end"),
|
|
) != (
|
|
observation["source_id"],
|
|
observation["period_start"],
|
|
observation["period_end"],
|
|
):
|
|
raise PriceSourceError("price manifest previous snapshot scope mismatch")
|
|
previous_snapshot = load_verified_snapshot(previous_snapshot_id)
|
|
previous_source = previous_snapshot.get("source")
|
|
if not isinstance(previous_source, dict):
|
|
raise PriceSourceError("price manifest previous snapshot source metadata is invalid")
|
|
for field in ("source_id", "period_start", "period_end", "raw_payload_hash"):
|
|
if previous_source.get(field) != previous_entry.get(field):
|
|
raise PriceSourceError("price manifest previous snapshot metadata mismatch")
|
|
|
|
expected_diff = (
|
|
_snapshot_diff(
|
|
previous_snapshot,
|
|
current_snapshot,
|
|
raw_changed=observation["raw_payload_hash"] != observation["previous_raw_payload_hash"],
|
|
)
|
|
if previous_snapshot is not None
|
|
else _snapshot_diff(current_snapshot, current_snapshot, raw_changed=None)
|
|
)
|
|
if observation["diff"] != expected_diff:
|
|
raise PriceSourceError("price manifest observation diff validation failed")
|
|
expected_status = "initial" if previous_snapshot is None else ("revised" if expected_diff["normalized_content_changed"] else "unchanged")
|
|
if observation["revision_status"] != expected_status:
|
|
raise PriceSourceError("price manifest observation revision status mismatch")
|
|
|
|
def _snapshot_path(self, snapshot_id: str) -> Path:
|
|
if not isinstance(snapshot_id, str) or not _SNAPSHOT_ID_RE.fullmatch(snapshot_id):
|
|
raise PriceSourceError("invalid price snapshot id")
|
|
candidate = (self.snapshot_dir / f"{snapshot_id}.json").resolve()
|
|
if candidate.parent != self.snapshot_dir.resolve():
|
|
raise PriceSourceError("price snapshot path escaped store root")
|
|
return candidate
|
|
|
|
def load_snapshot(self, snapshot_id: str) -> dict[str, Any]:
|
|
manifest = self.load_manifest()
|
|
return self._load_snapshot(snapshot_id, manifest)
|
|
|
|
def _load_snapshot(self, snapshot_id: str, manifest: Mapping[str, Any]) -> dict[str, Any]:
|
|
snapshot_path = self._snapshot_path(snapshot_id)
|
|
if not snapshot_path.is_file():
|
|
raise FileNotFoundError(snapshot_id)
|
|
try:
|
|
snapshot = json.loads(snapshot_path.read_text(encoding="utf-8"))
|
|
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
|
|
raise PriceSourceError("price snapshot is unreadable") from exc
|
|
if not isinstance(snapshot, dict):
|
|
raise PriceSourceError("price snapshot must be an object")
|
|
source = snapshot.get("source")
|
|
if not isinstance(source, dict) or source.get("snapshot_id") != snapshot_id:
|
|
raise PriceSourceError("price snapshot identity mismatch")
|
|
if not isinstance(source.get("point_in_time"), bool):
|
|
raise PriceSourceError("price snapshot metadata point_in_time must be boolean")
|
|
raw_files = source.get("raw_payload_files")
|
|
if not isinstance(raw_files, dict) or not raw_files:
|
|
raise PriceSourceError("price snapshot raw payload manifest is missing")
|
|
raw_payloads: dict[str, bytes] = {}
|
|
raw_root = (self.raw_dir / snapshot_id).resolve()
|
|
if raw_root.parent != self.raw_dir.resolve():
|
|
raise PriceSourceError("price raw path escaped store root")
|
|
for provider_symbol, filename in raw_files.items():
|
|
if _raw_filename(str(provider_symbol)) != filename:
|
|
raise PriceSourceError("price raw filename mismatch")
|
|
raw_path = (raw_root / filename).resolve()
|
|
if raw_path.parent != raw_root or not raw_path.is_file():
|
|
raise PriceSourceError("price raw payload is missing")
|
|
raw_payloads[str(provider_symbol)] = raw_path.read_bytes()
|
|
actual_hash = _combined_hash(raw_payloads)
|
|
if actual_hash != source.get("raw_payload_hash"):
|
|
raise PriceSourceError("price raw payload hash mismatch")
|
|
if source.get("point_in_time") is True:
|
|
_validate_pit_archive_snapshot(snapshot)
|
|
else:
|
|
_validate_non_pit_snapshot_shape(snapshot)
|
|
_validate_series_raw_lineage(snapshot, raw_payloads)
|
|
declared_snapshot_hash = source.get("normalized_snapshot_hash")
|
|
if source.get("normalized_hash_algorithm") != NORMALIZED_HASH_ALGORITHM or not isinstance(declared_snapshot_hash, str) or _normalized_snapshot_hash(snapshot) != declared_snapshot_hash:
|
|
raise PriceSourceError("price normalized snapshot hash mismatch")
|
|
manifest_entry = manifest.get("snapshots", {}).get(snapshot_id)
|
|
if not isinstance(manifest_entry, dict):
|
|
raise PriceSourceError("price manifest entry is missing")
|
|
if manifest_entry.get("snapshot_id") != snapshot_id or manifest_entry.get("snapshot_file") != snapshot_path.name:
|
|
raise PriceSourceError("price manifest metadata mismatch")
|
|
if manifest_entry.get("source_id") != source.get("source_id") or manifest_entry.get("quality") != source.get("quality"):
|
|
raise PriceSourceError("price manifest metadata mismatch")
|
|
if manifest_entry.get("point_in_time") is not source.get("point_in_time"):
|
|
raise PriceSourceError("price manifest metadata mismatch")
|
|
if manifest_entry.get("raw_payload_hash") != actual_hash or manifest_entry.get("normalized_snapshot_hash") != declared_snapshot_hash or manifest_entry.get("normalized_hash_algorithm") != NORMALIZED_HASH_ALGORITHM:
|
|
raise PriceSourceError("price manifest hash mismatch")
|
|
metadata_error = "price manifest point-in-time metadata mismatch" if source.get("point_in_time") is True else "price manifest metadata mismatch"
|
|
hardened_entry = manifest_entry.get("observation_history_contract") == OBSERVATION_HISTORY_CONTRACT
|
|
for field in (
|
|
"source_url",
|
|
"retrieved_at",
|
|
"period_start",
|
|
"period_end",
|
|
"parser_version",
|
|
"archive_contract",
|
|
"provider_release_id",
|
|
"point_in_time_evidence",
|
|
"adjusted_prices",
|
|
"return_price_field",
|
|
"snapshot_file",
|
|
):
|
|
if (field in manifest_entry or hardened_entry) and manifest_entry.get(field) != source.get(field):
|
|
raise PriceSourceError(metadata_error)
|
|
if ("symbols" in manifest_entry or hardened_entry) and manifest_entry.get("symbols") != sorted(snapshot.get("series", {}).keys()):
|
|
raise PriceSourceError(metadata_error)
|
|
return snapshot
|
|
|
|
def list_snapshots(self) -> list[dict[str, Any]]:
|
|
entries = list(self.load_manifest().get("snapshots", {}).values())
|
|
return sorted(
|
|
entries,
|
|
key=lambda item: (
|
|
_parse_timestamp(item.get("last_seen_at", item.get("retrieved_at")), "snapshot retrieved_at"),
|
|
str(item.get("snapshot_id", "")),
|
|
),
|
|
)
|
|
|
|
def latest_snapshot_entry(self) -> dict[str, Any] | None:
|
|
entries = self.list_snapshots()
|
|
return entries[-1] if entries else None
|
|
|
|
def list_observations(self) -> list[dict[str, Any]]:
|
|
observations = list(self.load_manifest().get("observations", []))
|
|
return sorted(observations, key=_observation_order_key)
|
|
|
|
def _validate_observation_append_order(
|
|
self,
|
|
manifest: Mapping[str, Any],
|
|
snapshot: Mapping[str, Any],
|
|
*,
|
|
raw_hash: str,
|
|
snapshot_id: str,
|
|
) -> dict[str, Any] | None:
|
|
source = snapshot.get("source")
|
|
if not isinstance(source, dict):
|
|
raise PriceSourceError("price snapshot source metadata is required")
|
|
retrieved_at = source.get("retrieved_at")
|
|
if not isinstance(retrieved_at, str) or not retrieved_at.strip():
|
|
raise PriceSourceError("price snapshot retrieved_at is required")
|
|
current_base = (_parse_timestamp(retrieved_at, "observation retrieved_at"), snapshot_id)
|
|
scope = (
|
|
str(source.get("source_id") or ""),
|
|
str(source.get("period_start") or ""),
|
|
str(source.get("period_end") or ""),
|
|
)
|
|
observations = manifest.get("observations", [])
|
|
matching = [
|
|
item
|
|
for item in observations
|
|
if isinstance(item, dict)
|
|
and (
|
|
str(item.get("source_id") or ""),
|
|
str(item.get("period_start") or ""),
|
|
str(item.get("period_end") or ""),
|
|
)
|
|
== scope
|
|
]
|
|
for item in matching:
|
|
if (
|
|
item.get("snapshot_id") == snapshot_id
|
|
and item.get("retrieved_at") == retrieved_at
|
|
and item.get("raw_payload_hash") == raw_hash
|
|
):
|
|
return item
|
|
latest = max(matching, key=_observation_order_key, default=None)
|
|
if latest is not None:
|
|
latest_base = _observation_order_base(latest)
|
|
if current_base < latest_base:
|
|
raise PriceSourceError("cannot append out-of-order observation with an immutable predecessor chain")
|
|
if current_base == latest_base:
|
|
raise PriceSourceError("cannot append ambiguous observation with an immutable predecessor chain")
|
|
return None
|
|
|
|
def _append_observation(
|
|
self,
|
|
manifest: dict[str, Any],
|
|
snapshot: Mapping[str, Any],
|
|
*,
|
|
raw_hash: str,
|
|
snapshot_id: str,
|
|
) -> dict[str, Any] | None:
|
|
source = snapshot.get("source")
|
|
if not isinstance(source, dict):
|
|
return None
|
|
retrieved_at = source.get("retrieved_at")
|
|
if not isinstance(retrieved_at, str) or not retrieved_at.strip():
|
|
return None
|
|
_parse_timestamp(retrieved_at, "observation retrieved_at")
|
|
duplicate = self._validate_observation_append_order(
|
|
manifest,
|
|
snapshot,
|
|
raw_hash=raw_hash,
|
|
snapshot_id=snapshot_id,
|
|
)
|
|
if duplicate is not None:
|
|
return duplicate
|
|
scope = (
|
|
str(source.get("source_id") or ""),
|
|
str(source.get("period_start") or ""),
|
|
str(source.get("period_end") or ""),
|
|
)
|
|
observations = manifest.setdefault("observations", [])
|
|
matching = [
|
|
item
|
|
for item in observations
|
|
if isinstance(item, dict)
|
|
and (
|
|
str(item.get("source_id") or ""),
|
|
str(item.get("period_start") or ""),
|
|
str(item.get("period_end") or ""),
|
|
)
|
|
== scope
|
|
]
|
|
current_base = (_parse_timestamp(retrieved_at, "observation retrieved_at"), snapshot_id)
|
|
previous = max(
|
|
(
|
|
item
|
|
for item in matching
|
|
if _observation_order_base(item) < current_base
|
|
),
|
|
key=_observation_order_key,
|
|
default=None,
|
|
)
|
|
previous_snapshot: dict[str, Any] | None = None
|
|
previous_raw_hash: str | None = None
|
|
previous_snapshot_id: str | None = None
|
|
if previous is not None:
|
|
previous_snapshot_id = str(previous.get("snapshot_id") or "") or None
|
|
previous_raw_hash = str(previous.get("raw_payload_hash") or "") or None
|
|
if previous_snapshot_id:
|
|
try:
|
|
previous_snapshot = self.load_snapshot(previous_snapshot_id)
|
|
except (FileNotFoundError, PriceSourceError) as exc:
|
|
raise PriceSourceError("previous price snapshot integrity validation failed") from exc
|
|
diff = (
|
|
_snapshot_diff(
|
|
previous_snapshot,
|
|
snapshot,
|
|
raw_changed=previous_raw_hash != raw_hash,
|
|
)
|
|
if previous_snapshot is not None
|
|
else _snapshot_diff(snapshot, snapshot, raw_changed=None)
|
|
)
|
|
if previous_snapshot is not None and previous_raw_hash == raw_hash and diff["normalized_content_changed"]:
|
|
raise PriceSourceError("cannot change normalized content for an unchanged raw payload")
|
|
revision_status = "initial" if previous is None else ("revised" if diff["normalized_content_changed"] else "unchanged")
|
|
observation_without_id = {
|
|
"snapshot_id": snapshot_id,
|
|
"source_id": source.get("source_id"),
|
|
"period_start": source.get("period_start"),
|
|
"period_end": source.get("period_end"),
|
|
"retrieved_at": retrieved_at,
|
|
"raw_payload_hash": raw_hash,
|
|
"previous_snapshot_id": previous_snapshot_id,
|
|
"previous_raw_payload_hash": previous_raw_hash,
|
|
"revision_status": revision_status,
|
|
"diff": diff,
|
|
}
|
|
observation_id = _observation_id(observation_without_id)
|
|
observation = {"observation_id": observation_id, **observation_without_id}
|
|
_validate_observation_entry(observation)
|
|
if not any(isinstance(item, dict) and item.get("observation_id") == observation_id for item in observations):
|
|
observations.append(observation)
|
|
|
|
manifest_entry = manifest["snapshots"].get(snapshot_id)
|
|
if isinstance(manifest_entry, dict):
|
|
first_seen = str(manifest_entry.get("first_seen_at") or retrieved_at)
|
|
manifest_entry["first_seen_at"] = _earliest_timestamp([first_seen, retrieved_at])
|
|
last_seen = str(manifest_entry.get("last_seen_at") or retrieved_at)
|
|
manifest_entry["last_seen_at"] = _latest_timestamp([last_seen, retrieved_at])
|
|
manifest_entry["observation_count"] = sum(
|
|
1 for item in observations if isinstance(item, dict) and item.get("snapshot_id") == snapshot_id
|
|
)
|
|
return observation
|
|
|
|
def persist(self, snapshot: dict[str, Any], raw_payloads: Mapping[str, bytes]) -> dict[str, Any]:
|
|
source = snapshot.get("source")
|
|
if not isinstance(source, dict) or not source.get("snapshot_id"):
|
|
raise PriceSourceError("price snapshot source metadata is required")
|
|
if not raw_payloads or any(not isinstance(symbol, str) or not _PROVIDER_SYMBOL_RE.fullmatch(symbol) for symbol in raw_payloads) or any(not isinstance(value, bytes) or not value for value in raw_payloads.values()):
|
|
raise PriceSourceError("price raw payloads must be non-empty bytes")
|
|
raw_hash = _combined_hash(raw_payloads)
|
|
if source.get("raw_payload_hash") and source["raw_payload_hash"] != raw_hash:
|
|
raise PriceSourceError("raw payload hash does not match price snapshot")
|
|
if "point_in_time" in source and not isinstance(source["point_in_time"], bool):
|
|
raise PriceSourceError("price snapshot metadata point_in_time must be boolean")
|
|
stored = copy.deepcopy(snapshot)
|
|
stored_source = stored["source"]
|
|
snapshot_id = str(stored_source["snapshot_id"])
|
|
if not _SNAPSHOT_ID_RE.fullmatch(snapshot_id):
|
|
raise PriceSourceError("invalid price snapshot id")
|
|
stored_source["raw_payload_hash"] = raw_hash
|
|
stored_source["raw_payload_files"] = {symbol: _raw_filename(symbol) for symbol in raw_payloads}
|
|
stored_source["snapshot_file"] = f"{snapshot_id}.json"
|
|
stored_source["normalized_hash_algorithm"] = NORMALIZED_HASH_ALGORITHM
|
|
stored_source["normalized_snapshot_hash"] = _normalized_snapshot_hash(stored)
|
|
manifest = self.load_manifest()
|
|
if stored_source.get("point_in_time") is True:
|
|
_validate_pit_archive_snapshot(stored)
|
|
else:
|
|
_validate_non_pit_snapshot_shape(stored)
|
|
_validate_series_raw_lineage(stored, raw_payloads)
|
|
existing_entry = manifest["snapshots"].get(snapshot_id)
|
|
existing_snapshot: dict[str, Any] | None = None
|
|
if existing_entry is not None:
|
|
if not isinstance(existing_entry, dict):
|
|
raise PriceSourceError("price manifest entry is invalid")
|
|
if existing_entry.get("raw_payload_hash") != raw_hash:
|
|
raise PriceSourceError("cannot overwrite immutable price snapshot")
|
|
existing_snapshot = self.load_snapshot(snapshot_id)
|
|
if _revision_comparison_payload(existing_snapshot) != _revision_comparison_payload(stored):
|
|
raise PriceSourceError("cannot overwrite immutable price snapshot")
|
|
staged_manifest = copy.deepcopy(manifest)
|
|
if existing_entry is None:
|
|
staged_manifest["snapshots"][snapshot_id] = {
|
|
"snapshot_id": snapshot_id,
|
|
"source_id": stored_source.get("source_id"),
|
|
"retrieved_at": stored_source.get("retrieved_at"),
|
|
"period_start": stored_source.get("period_start"),
|
|
"period_end": stored_source.get("period_end"),
|
|
"raw_payload_hash": raw_hash,
|
|
"normalized_snapshot_hash": stored_source["normalized_snapshot_hash"],
|
|
"normalized_hash_algorithm": NORMALIZED_HASH_ALGORITHM,
|
|
"parser_version": stored_source.get("parser_version"),
|
|
"quality": stored_source.get("quality"),
|
|
"point_in_time": stored_source.get("point_in_time"),
|
|
"archive_contract": stored_source.get("archive_contract"),
|
|
"provider_release_id": stored_source.get("provider_release_id"),
|
|
"source_url": stored_source.get("source_url"),
|
|
"point_in_time_evidence": stored_source.get("point_in_time_evidence"),
|
|
"adjusted_prices": stored_source.get("adjusted_prices"),
|
|
"return_price_field": stored_source.get("return_price_field"),
|
|
"symbols": sorted(stored.get("series", {}).keys()),
|
|
"snapshot_file": stored_source["snapshot_file"],
|
|
"observation_history_contract": OBSERVATION_HISTORY_CONTRACT,
|
|
}
|
|
self._append_observation(staged_manifest, stored, raw_hash=raw_hash, snapshot_id=snapshot_id)
|
|
self._validate_observation_history(staged_manifest, snapshot_overrides={snapshot_id: stored})
|
|
if existing_entry is None:
|
|
for provider_symbol, payload in raw_payloads.items():
|
|
_atomic_write(self.raw_dir / snapshot_id / _raw_filename(provider_symbol), payload)
|
|
_atomic_write(self.snapshot_dir / stored_source["snapshot_file"], (json.dumps(stored, ensure_ascii=False, indent=2, sort_keys=True) + "\n").encode("utf-8"))
|
|
_atomic_write(self.manifest_path, (json.dumps(staged_manifest, ensure_ascii=False, indent=2, sort_keys=True) + "\n").encode("utf-8"))
|
|
return existing_snapshot if existing_snapshot is not None else stored
|
|
|
|
|
|
def collect_price_snapshot(
|
|
root: Path | str,
|
|
*,
|
|
start: str,
|
|
end: str,
|
|
symbol_map: Mapping[str, str] = DEFAULT_SYMBOL_MAP,
|
|
provider: YahooChartProvider | Any | None = None,
|
|
retrieved_at: str | None = None,
|
|
) -> dict[str, Any]:
|
|
if not symbol_map:
|
|
raise PriceSourceError("symbol_map must not be empty")
|
|
provider = provider or YahooChartProvider()
|
|
normalized: dict[str, Any] = {}
|
|
raw_payloads: dict[str, bytes] = {}
|
|
for canonical_symbol, provider_symbol in symbol_map.items():
|
|
series, raw_bytes = provider.fetch_series(canonical_symbol, provider_symbol, start, end)
|
|
normalized[canonical_symbol] = series
|
|
raw_payloads[provider_symbol] = raw_bytes
|
|
raw_hash = _combined_hash(raw_payloads)
|
|
retrieved = retrieved_at or datetime.now(timezone.utc).isoformat()
|
|
snapshot_id = f"prices-yahoo-chart-{start}-{end}-{raw_hash[:12]}"
|
|
source = {
|
|
"source_id": "yahoo.chart",
|
|
"source_url": "https://query1.finance.yahoo.com/v8/finance/chart/{provider_symbol}",
|
|
"retrieved_at": retrieved,
|
|
"period_start": start,
|
|
"period_end": end,
|
|
"raw_payload_hash": raw_hash,
|
|
"parser_version": PRICE_PARSER_VERSION,
|
|
"quality": "revised_vendor_history",
|
|
"point_in_time": False,
|
|
"adjusted_prices": True,
|
|
"return_price_field": "close",
|
|
"snapshot_id": snapshot_id,
|
|
"bar_counts": {symbol: len(series["bars"]) for symbol, series in normalized.items()},
|
|
}
|
|
snapshot = {"schema_version": PRICE_SCHEMA_VERSION, "source": source, "series": normalized, "benchmark_symbol": "SET50" if "SET50" in normalized else None}
|
|
return PriceSnapshotStore(root).persist(snapshot, raw_payloads)
|