1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
|
"""Normalization tests — synthetic fixtures only (_synthetic)."""
import math
import sys
import types
from pathlib import Path
import pandas as pd
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
sys.path.insert(0, str(Path(__file__).resolve().parent))
from fixtures_synth import synthetic_daily, SYNTH # noqa: E402
from worker.normalize import normalize_frame, build_manifest_entry, \
compute_object_hash, coverage_summary, NormalizeError # noqa: E402
def test_synthetic_marker_present():
assert SYNTH["_synthetic"] is True
df = synthetic_daily()
assert df.attrs["synthetic"] is True
def test_normalize_keeps_raw_extra_fields_and_precision():
df = synthetic_daily()
out = normalize_frame(df, symbol="SH#600000")
assert "_synthetic" not in out.columns
# canonical columns present
for col in ("date", "open", "high", "low", "close", "volume"):
assert col in out.columns
# raw extra fields preserved
assert "amount" in out.columns and "turnover" in out.columns
# precision kept, no rounding drift
assert out["close"].iloc[0] == df["close"].iloc[0]
assert out["volume"].iloc[3] == df["volume"].iloc[3]
# symbol never substituted across instruments
assert out["symbol"].iloc[5] == "SH#600000"
assert out["date"].notna().all()
def test_normalize_rejects_missing_ohlcv():
df = synthetic_daily()
df.loc[4, "close"] = math.nan
with pytest.raises(NormalizeError):
normalize_frame(df, symbol="SH#600000")
def test_normalize_rejects_bad_dates():
df = synthetic_daily()
df["date"] = df["date"].astype(object)
df.loc[2, "date"] = "not-a-date"
with pytest.raises(NormalizeError):
normalize_frame(df, symbol="SH#600000")
def test_normalize_sorts_and_dedups_dates():
df = synthetic_daily()
df = pd.concat([df.iloc[5:], df.iloc[:5]]).reset_index(drop=True)
out = normalize_frame(df, symbol="SH#600000")
assert list(out["date"]) == sorted(out["date"])
assert len(out) == 20
def test_object_hash_stable_and_content_sensitive():
df = synthetic_daily()
h1 = compute_object_hash(df)
h2 = compute_object_hash(normalize_frame(df, symbol="SH#600000"))
assert h1 == h2 and len(h1) == 64
df2 = synthetic_daily(base=11.0)
assert compute_object_hash(df2) != h1
def test_coverage_summary_and_manifest_entry():
df = synthetic_daily(days=10, extra_fields=False)
cov = coverage_summary(df)
assert cov["actual_start"] == df["date"].iloc[0]
entry = build_manifest_entry(
df, instrument={"symbol": "SH#600000", "market": "cn", "asset_type": "stock"},
provider="akshare", endpoint="stock_zh_a_hist", params={"period": "daily"},
adjustment="none", requested_start="2024-01-01", requested_end="2024-12-31",
raw_object_hash="rawhash", warnings=[],
schema_version="1", normalization_version="1",
path="objects/ab/cd.json", fetched_at="2026-09-16T00:00:00Z",
akshare_version="1.18.95",
)
assert entry["immutable"] is True
assert entry["row_count"] == 10
assert entry["adjustment"] == "none"
assert entry["actual_start"] <= entry["actual_end"]
assert "potential symbol substitution" not in entry["warnings"]
for internal in ("path",):
assert internal in entry # server strips internal-only fields downstream
|
