1 files changed, 160 insertions, 0 deletions
diff --git a/worker/normalize.py b/worker/normalize.py
new file mode 100644
index 0000000..4cb373d
--- /dev/null
+++ b/worker/normalize.py
@@ -0,0 +1,160 @@
+"""Normalization: provider frames -> canonical OHLCV schema + content hashes."""
+from __future__ import annotations
+
+import hashlib
+import json
+from typing import Any
+
+import pandas as pd
+
+SCHEMA_VERSION = "1"
+NORMALIZATION_VERSION = "1"
+
+# Canonical instrument identity format: "<EXCHANGE>#<CODE>", e.g. SH#600000.
+# Prices are never substituted across canonical symbols.
+
+# AKShare (Chinese) -> canonical column names. Non-mapped fields are preserved
+# verbatim as raw provider fields (units/precision untouched).
+COLUMN_MAP = {
+    "日期": "date",
+    "开盘": "open",
+    "收盘": "close",
+    "最高": "high",
+    "最低": "low",
+    "成交量": "volume",
+}
+
+REQUIRED = ["date", "open", "high", "low", "close", "volume"]
+
+
+class NormalizeError(ValueError):
+    def __init__(self, code: str, message: str, details: Any = None):
+        super().__init__(message)
+        self.code = code
+        self.details = details
+
+
+def map_columns(df: pd.DataFrame) -> pd.DataFrame:
+    return df.rename(columns=COLUMN_MAP)
+
+
+def normalize_frame(df: pd.DataFrame, symbol: str) -> pd.DataFrame:
+    if df is None or df.empty:
+        raise NormalizeError("empty_response", "provider returned no rows")
+    missing_req = {c for c in REQUIRED if c not in df.columns}
+    if missing_req:
+        raise NormalizeError(
+            "missing_columns",
+            f"provider frame missing standard columns: {sorted(missing_req)}",
+            {"columns": list(df.columns)},
+        )
+    out = df.copy()
+    out["date"] = pd.to_datetime(out["date"], errors="coerce")
+    if out["date"].isna().any():
+        bad = out.loc[out["date"].isna()].index.tolist()
+        raise NormalizeError(
+            "bad_date", "unparseable date values", {"rows": [int(i) for i in bad[:5]]}
+        )
+    for col in ("open", "high", "low", "close", "volume"):
+        out[col] = pd.to_numeric(out[col], errors="coerce")
+        if out[col].isna().any():
+            rows = out.loc[out[col].isna()].index.tolist()
+            first = out.loc[rows[0], "date"].date().isoformat()
+            raise NormalizeError(
+                "missing_ohlcv",
+                f"missing or non-numeric '{col}' on {first} — gap kept, never imputed",
+                {"column": col, "row_count": len(rows), "rows": [int(i) for i in rows[:5]]},
+            )
+    out["date"] = out["date"].dt.strftime("%Y-%m-%d")
+    out = out.sort_values("date", kind="mergesort")
+    out = out.drop_duplicates(subset="date", keep="last")
+    # canonical identity, prices never substituted across symbols
+    out["symbol"] = symbol
+    return out.reset_index(drop=True)
+
+
+def frame_records(df: pd.DataFrame) -> list[dict]:
+    return json.loads(df.to_json(orient="records", force_ascii=False))
+
+
+def compute_object_hash(df: pd.DataFrame) -> str:
+    """Content hash over canonical records; independent of request/user ids."""
+    recs = frame_records(df)
+    return hashlib.sha256(
+        json.dumps(recs, sort_keys=True, ensure_ascii=False).encode("utf-8")
+    ).hexdigest()
+
+
+def coverage_summary(df_or_dates) -> dict:
+    if isinstance(df_or_dates, pd.DataFrame):
+        dates = pd.to_datetime(df_or_dates["date"]).tolist()
+    else:
+        dates = pd.to_datetime(list(df_or_dates)).tolist()
+    if not dates:
+        return {"actual_start": None, "actual_end": None, "segments": [], "gaps": []}
+    segs = [[dates[0], dates[0]]]
+    gaps: list[dict] = []
+    for prev, cur in zip(dates, dates[1:]):
+        delta = cur - prev
+        if delta.days == 1:
+            segs[-1][1] = cur
+        else:
+            segs.append([cur, cur])
+            gaps.append({"after": prev.date().isoformat(), "before": cur.date().isoformat(),
+                         "calendar_days": delta.days - 1})
+    fmt = lambda d: d.date().isoformat()
+    return {
+        "actual_start": fmt(dates[0]),
+        "actual_end": fmt(dates[-1]),
+        "segments": [[fmt(a), fmt(b)] for a, b in segs],
+        "gaps": gaps,
+    }
+
+
+def build_manifest_entry(
+    df: pd.DataFrame,
+    *,
+    instrument: dict,
+    provider: str,
+    endpoint: str,
+    params: dict,
+    adjustment: str,
+    requested_start: str,
+    requested_end: str,
+    raw_object_hash: str,
+    warnings: list,
+    schema_version: str = SCHEMA_VERSION,
+    normalization_version: str = NORMALIZATION_VERSION,
+    path: str,
+    fetched_at: str,
+    akshare_version: str,
+) -> dict:
+    cov = coverage_summary(df)
+    return {
+        "instrument": {
+            "symbol": instrument["symbol"],
+            "market": instrument["market"],
+            "asset_type": instrument["asset_type"],
+            **({"name": instrument["name"]} if instrument.get("name") else {}),
+        },
+        "object_hash": compute_object_hash(df),
+        "path": path,  # internal-only: backend rewrites before client exposure
+        "raw_object_hash": raw_object_hash,
+        "provider": provider,
+        "endpoint": endpoint,
+        "params": params,
+        "akshare_version": akshare_version,
+        "fetched_at": fetched_at,
+        "schema_version": schema_version,
+        "normalization_version": normalization_version,
+        "adjustment": adjustment,
+        "requested_start": requested_start,
+        "requested_end": requested_end,
+        "actual_start": cov["actual_start"],
+        "actual_end": cov["actual_end"],
+        "coverage": {"segments": cov["segments"], "gaps": cov["gaps"]},
+        "row_count": int(len(df)),
+        "columns": list(df.columns),
+        "warnings": list(warnings),
+        "immutable": True,
+    }