From 2e492b375a656ec3f9a28da2a5e6e275472f37ab Mon Sep 17 00:00:00 2001 From: Kunthawat Greethong Date: Tue, 25 Aug 2026 17:04:06 +0700 Subject: [PATCH] [verified] Extend price universe to full SET50 (49 symbols + index) so simulation allocates across all names MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- backend/app/prices.py | 928 +++++++++++++++++++++++++++++++++++++++--- 1 file changed, 881 insertions(+), 47 deletions(-) diff --git a/backend/app/prices.py b/backend/app/prices.py index 998c874..9e2c73c 100644 --- a/backend/app/prices.py +++ b/backend/app/prices.py @@ -28,13 +28,65 @@ DEFAULT_SYMBOL_MAP = { "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): @@ -48,6 +100,114 @@ def _parse_date(value: str) -> date: 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): @@ -67,6 +227,168 @@ def _normalized_snapshot_hash(snapshot: dict[str, Any]) -> str: 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") @@ -74,6 +396,116 @@ def _raw_filename(provider_symbol: str) -> str: 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") @@ -81,6 +513,18 @@ def _atomic_write(path: Path, content: bytes) -> None: 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], *, @@ -89,6 +533,8 @@ def normalize_yahoo_chart( 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] @@ -98,7 +544,7 @@ def normalize_yahoo_chart( 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 timestamps or not isinstance(adjusted, list): + 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}") @@ -123,17 +569,15 @@ def normalize_yahoo_chart( raw_adjusted_close = at(adjusted, index) if timestamp is None or raw_close is None or raw_adjusted_close is None: continue - try: - timestamp_value = float(timestamp) - close = float(raw_close) - adjusted_close = float(raw_adjusted_close) - except (KeyError, IndexError, TypeError, ValueError) as exc: - raise PriceSourceError(f"invalid Yahoo price row for {provider_symbol}") from exc - if not math.isfinite(timestamp_value) or not math.isfinite(close) or not math.isfinite(adjusted_close): - 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}") - trading_date = datetime.fromtimestamp(timestamp_value, tz=timezone.utc).astimezone(market_zone).date().isoformat() + 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) @@ -148,13 +592,13 @@ def normalize_yahoo_chart( } for key in ("open", "high", "low"): if bar[key] is not None: - value = float(bar[key]) - if not math.isfinite(value) or value <= 0: + 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 = float(bar["volume"]) - if not math.isfinite(volume) or volume < 0: + 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) @@ -234,14 +678,187 @@ class PriceSnapshotStore: return {"schema_version": PRICE_SCHEMA_VERSION, "snapshots": {}} try: manifest = json.loads(self.manifest_path.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError) as exc: + except (OSError, UnicodeError, json.JSONDecodeError) as exc: raise PriceSourceError("price manifest is unreadable") from exc - if manifest.get("schema_version") != PRICE_SCHEMA_VERSION or not isinstance(manifest.get("snapshots"), dict): + 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 re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}", snapshot_id): + 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(): @@ -249,21 +866,24 @@ class PriceSnapshotStore: 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, json.JSONDecodeError) as exc: + 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") - 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") 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") @@ -281,30 +901,219 @@ class PriceSnapshotStore: actual_hash = _combined_hash(raw_payloads) if actual_hash != source.get("raw_payload_hash"): raise PriceSourceError("price raw payload hash mismatch") - manifest_entry = self.load_manifest().get("snapshots", {}).get(snapshot_id) + 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: (str(item.get("retrieved_at", "")), str(item.get("snapshot_id", "")))) + 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(value, bytes) or not value for value in raw_payloads.values()): + 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: @@ -314,36 +1123,61 @@ class PriceSnapshotStore: stored = copy.deepcopy(snapshot) stored_source = stored["source"] snapshot_id = str(stored_source["snapshot_id"]) - if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}", 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) - 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")) manifest = self.load_manifest() - 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"), - "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"], - } - _atomic_write(self.manifest_path, (json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True) + "\n").encode("utf-8")) - return stored + 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(