From 0cf9920e2ddd86d2cd90e932fc5d8e88cf1368f4 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 07:18:12 +0800 Subject: [PATCH 1/8] fix: use reliable ETF history source Co-Authored-By: Codex --- .github/workflows/publish-market-history.yml | 3 +- pyproject.toml | 2 +- .../akshare_market_history.py | 84 ++++++++++++++++++- tests/test_akshare_staging.py | 22 +++++ tests/test_publish_market_history_workflow.py | 1 + 5 files changed, 106 insertions(+), 6 deletions(-) diff --git a/.github/workflows/publish-market-history.yml b/.github/workflows/publish-market-history.yml index bb67404..febdf2d 100644 --- a/.github/workflows/publish-market-history.yml +++ b/.github/workflows/publish-market-history.yml @@ -31,12 +31,13 @@ jobs: set -euo pipefail python -m pip install --upgrade pip python -m pip install --no-deps -e . - python -m pip install pandas "akshare>=1.14" + python -m pip install pandas "requests>=2.31" - name: Fetch complete ETF history run: | set -euo pipefail cneq-stage-akshare-market-history \ + --source yahoo \ --start-date 20220101 \ --output "${RUNNER_TEMP}/cn-market-history/cn_etf_market_history.csv" diff --git a/pyproject.toml b/pyproject.toml index ae09b77..d9f47f8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,7 +15,7 @@ dependencies = [ [project.optional-dependencies] test = ["pytest>=8"] -public-data = ["akshare>=1.14"] +public-data = ["akshare>=1.14", "requests>=2.31"] [project.scripts] cneq-build-dividend-quality-snapshot = "cn_equity_snapshot_pipelines.dividend_quality:main" diff --git a/src/cn_equity_snapshot_pipelines/akshare_market_history.py b/src/cn_equity_snapshot_pipelines/akshare_market_history.py index 61213da..2eabf94 100644 --- a/src/cn_equity_snapshot_pipelines/akshare_market_history.py +++ b/src/cn_equity_snapshot_pipelines/akshare_market_history.py @@ -2,7 +2,8 @@ import argparse import time -from datetime import datetime, timezone +import urllib.parse +from datetime import datetime, timedelta, timezone from pathlib import Path import pandas as pd @@ -45,6 +46,70 @@ def _import_akshare(): return ak +def yahoo_symbol(value: object) -> str: + symbol = normalize_symbol(value) + suffix = ".SZ" if symbol.startswith(("0", "1", "3")) else ".SS" + return f"{symbol}{suffix}" + + +def fetch_yahoo_etf_history( + symbol: str, + *, + start_date: str = "20200101", + max_attempts: int = 3, + retry_delay_seconds: float = 1.0, +) -> pd.DataFrame: + import requests + + start = pd.Timestamp(start_date, tz="UTC") + end = pd.Timestamp(datetime.now(timezone.utc) + timedelta(days=1)) + query = urllib.parse.urlencode( + { + "period1": int(start.timestamp()), + "period2": int(end.timestamp()), + "interval": "1d", + "events": "history", + } + ) + url = f"https://query1.finance.yahoo.com/v8/finance/chart/{urllib.parse.quote(yahoo_symbol(symbol))}?{query}" + attempts = max(int(max_attempts), 1) + for attempt in range(1, attempts + 1): + try: + response = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=30) + response.raise_for_status() + payload = response.json() + result = payload.get("chart", {}).get("result") or [] + if not result: + raise ValueError(f"empty Yahoo ETF history for {symbol}") + series = result[0] + timestamps = series.get("timestamp") or [] + indicators = series.get("indicators") or {} + quotes = (indicators.get("quote") or [{}])[0] + adjusted = (indicators.get("adjclose") or [{}])[0].get("adjclose") or [] + closes = quotes.get("close") or [] + rows = [] + for index, raw_timestamp in enumerate(timestamps): + close = adjusted[index] if index < len(adjusted) else closes[index] if index < len(closes) else None + if close is None: + continue + rows.append( + { + "date": pd.Timestamp.fromtimestamp(int(raw_timestamp), tz="UTC").date().isoformat(), + "symbol": normalize_symbol(symbol), + "close": float(close), + } + ) + frame = pd.DataFrame(rows) + if frame.empty: + raise ValueError(f"empty Yahoo ETF history for {symbol}") + return frame + except Exception: + if attempt == attempts: + raise + time.sleep(max(float(retry_delay_seconds), 0.0) * attempt) + raise AssertionError("unreachable") + + def fetch_etf_history( symbol: str, *, @@ -88,13 +153,20 @@ def build_market_history_frame( ak=None, start_date: str = "20200101", request_delay_seconds: float = 0.5, + source: str = "akshare", ) -> pd.DataFrame: frames: list[pd.DataFrame] = [] errors: dict[str, str] = {} requested_symbols = tuple(dict.fromkeys(normalize_symbol(symbol) for symbol in symbols)) + if source not in {"akshare", "yahoo"}: + raise ValueError("source must be 'akshare' or 'yahoo'") + fetcher = fetch_etf_history if source == "akshare" else fetch_yahoo_etf_history for index, symbol in enumerate(requested_symbols): try: - frames.append(fetch_etf_history(symbol, ak=ak, start_date=start_date)) + kwargs = {"start_date": start_date} + if source == "akshare": + kwargs["ak"] = ak + frames.append(fetcher(symbol, **kwargs)) except Exception as exc: errors[normalize_symbol(symbol)] = str(exc) if index + 1 < len(requested_symbols): @@ -112,9 +184,10 @@ def write_market_history_csv( output_path: str | Path, symbols: tuple[str, ...] = DEFAULT_ETF_SYMBOLS, start_date: str = "20200101", + source: str = "akshare", ) -> dict[str, object]: - ak = _import_akshare() - frame = build_market_history_frame(symbols, ak=ak, start_date=start_date) + ak = _import_akshare() if source == "akshare" else None + frame = build_market_history_frame(symbols, ak=ak, start_date=start_date, source=source) path = Path(output_path) path.parent.mkdir(parents=True, exist_ok=True) frame.to_csv(path, index=False) @@ -123,6 +196,7 @@ def write_market_history_csv( "row_count": int(len(frame)), "symbols": sorted(frame["symbol"].unique().tolist()), "start_date": start_date, + "source": source, } @@ -131,12 +205,14 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--output", default="data/staging/market_history/etf_universe.latest.csv") parser.add_argument("--symbols", default=",".join(DEFAULT_ETF_SYMBOLS)) parser.add_argument("--start-date", default="20200101") + parser.add_argument("--source", choices=("akshare", "yahoo"), default="akshare") args = parser.parse_args(argv) symbols = tuple(symbol.strip() for symbol in args.symbols.split(",") if symbol.strip()) diagnostics = write_market_history_csv( output_path=args.output, symbols=symbols, start_date=args.start_date, + source=args.source, ) print(diagnostics) return 0 diff --git a/tests/test_akshare_staging.py b/tests/test_akshare_staging.py index 3a417dd..eedd4e0 100644 --- a/tests/test_akshare_staging.py +++ b/tests/test_akshare_staging.py @@ -122,3 +122,25 @@ def _fetch(symbol: str, **kwargs): with pytest.raises(RuntimeError, match="510500"): build_market_history_frame(("510300", "510500"), ak=object(), request_delay_seconds=0) + + +def test_build_market_history_frame_supports_yahoo_source(monkeypatch: pytest.MonkeyPatch): + from cn_equity_snapshot_pipelines import akshare_market_history as module + + monkeypatch.setattr( + module, + "fetch_yahoo_etf_history", + lambda symbol, **kwargs: pd.DataFrame( + {"date": ["2024-01-02"], "symbol": [symbol], "close": [10.0]} + ), + ) + + frame = build_market_history_frame( + ("510300", "159915"), + source="yahoo", + request_delay_seconds=0, + ) + + assert set(frame["symbol"]) == {"510300", "159915"} + assert module.yahoo_symbol("510300") == "510300.SS" + assert module.yahoo_symbol("159915") == "159915.SZ" diff --git a/tests/test_publish_market_history_workflow.py b/tests/test_publish_market_history_workflow.py index e9f1b99..d43e3b7 100644 --- a/tests/test_publish_market_history_workflow.py +++ b/tests/test_publish_market_history_workflow.py @@ -7,6 +7,7 @@ def test_publish_market_history_workflow_is_real_and_fail_closed() -> None: ).read_text(encoding="utf-8") assert "cneq-stage-akshare-market-history" in workflow + assert "--source yahoo" in workflow assert "--start-date 20220101" in workflow assert "cn_etf_market_history.csv" in workflow assert "actions/upload-artifact@v4" in workflow From 75c39798a333100b9138896ffd3769aabb945534 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 07:25:49 +0800 Subject: [PATCH 2/8] docs: define adjusted ETF price contract Co-Authored-By: Codex --- .../akshare_market_history.py | 3 ++ tests/test_akshare_staging.py | 32 +++++++++++++++++++ 2 files changed, 35 insertions(+) diff --git a/src/cn_equity_snapshot_pipelines/akshare_market_history.py b/src/cn_equity_snapshot_pipelines/akshare_market_history.py index 2eabf94..d53a0a6 100644 --- a/src/cn_equity_snapshot_pipelines/akshare_market_history.py +++ b/src/cn_equity_snapshot_pipelines/akshare_market_history.py @@ -31,6 +31,7 @@ "512690", "159928", ) +PRICE_BASIS = "adjusted_close" def normalize_symbol(value: object) -> str: @@ -85,6 +86,7 @@ def fetch_yahoo_etf_history( timestamps = series.get("timestamp") or [] indicators = series.get("indicators") or {} quotes = (indicators.get("quote") or [{}])[0] + # Preserve the existing AkShare adjust="qfq" contract: `close` is adjusted, not raw exchange close. adjusted = (indicators.get("adjclose") or [{}])[0].get("adjclose") or [] closes = quotes.get("close") or [] rows = [] @@ -197,6 +199,7 @@ def write_market_history_csv( "symbols": sorted(frame["symbol"].unique().tolist()), "start_date": start_date, "source": source, + "price_basis": PRICE_BASIS, } diff --git a/tests/test_akshare_staging.py b/tests/test_akshare_staging.py index eedd4e0..62c3a6a 100644 --- a/tests/test_akshare_staging.py +++ b/tests/test_akshare_staging.py @@ -144,3 +144,35 @@ def test_build_market_history_frame_supports_yahoo_source(monkeypatch: pytest.Mo assert set(frame["symbol"]) == {"510300", "159915"} assert module.yahoo_symbol("510300") == "510300.SS" assert module.yahoo_symbol("159915") == "159915.SZ" + + +def test_yahoo_history_preserves_adjusted_close_contract(monkeypatch: pytest.MonkeyPatch): + from cn_equity_snapshot_pipelines import akshare_market_history as module + + class _Response: + def raise_for_status(self) -> None: + return None + + def json(self) -> dict: + return { + "chart": { + "result": [ + { + "timestamp": [1704153600], + "indicators": { + "quote": [{"close": [12.0]}], + "adjclose": [{"adjclose": [10.0]}], + }, + } + ] + } + } + + import requests + + monkeypatch.setattr(requests, "get", lambda *args, **kwargs: _Response()) + + frame = module.fetch_yahoo_etf_history("510300", start_date="20240101") + + assert frame.iloc[0]["close"] == 10.0 + assert module.PRICE_BASIS == "adjusted_close" From 9d41e4ef61f24a905a44b6e5402ba11246e2e4ea Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 07:32:13 +0800 Subject: [PATCH 3/8] fix: use complete Tencent qfq history Co-Authored-By: Codex --- .github/workflows/publish-market-history.yml | 2 +- .../akshare_market_history.py | 69 +++++++++++++++++-- tests/test_akshare_staging.py | 2 + tests/test_publish_market_history_workflow.py | 2 +- 4 files changed, 69 insertions(+), 6 deletions(-) diff --git a/.github/workflows/publish-market-history.yml b/.github/workflows/publish-market-history.yml index febdf2d..c43323f 100644 --- a/.github/workflows/publish-market-history.yml +++ b/.github/workflows/publish-market-history.yml @@ -37,7 +37,7 @@ jobs: run: | set -euo pipefail cneq-stage-akshare-market-history \ - --source yahoo \ + --source tencent \ --start-date 20220101 \ --output "${RUNNER_TEMP}/cn-market-history/cn_etf_market_history.csv" diff --git a/src/cn_equity_snapshot_pipelines/akshare_market_history.py b/src/cn_equity_snapshot_pipelines/akshare_market_history.py index d53a0a6..58621ee 100644 --- a/src/cn_equity_snapshot_pipelines/akshare_market_history.py +++ b/src/cn_equity_snapshot_pipelines/akshare_market_history.py @@ -53,6 +53,62 @@ def yahoo_symbol(value: object) -> str: return f"{symbol}{suffix}" +def tencent_symbol(value: object) -> str: + symbol = normalize_symbol(value) + prefix = "sz" if symbol.startswith(("0", "1", "3")) else "sh" + return f"{prefix}{symbol}" + + +def fetch_tencent_etf_history( + symbol: str, + *, + start_date: str = "20200101", + max_attempts: int = 3, + retry_delay_seconds: float = 1.0, +) -> pd.DataFrame: + import requests + + start = pd.Timestamp(start_date) + end = pd.Timestamp(datetime.now(timezone.utc).date()) + rows: list[dict[str, object]] = [] + for first_year in range(start.year, end.year + 1, 2): + chunk_start = max(start, pd.Timestamp(first_year, 1, 1)) + chunk_end = min(end, pd.Timestamp(first_year + 1, 12, 31)) + params = { + "param": ( + f"{tencent_symbol(symbol)},day,{chunk_start.date().isoformat()}," + f"{chunk_end.date().isoformat()},2000,qfq" + ) + } + for attempt in range(1, max(int(max_attempts), 1) + 1): + try: + response = requests.get( + "https://web.ifzq.gtimg.cn/appstock/app/fqkline/get", + params=params, + headers={"User-Agent": "Mozilla/5.0"}, + timeout=30, + ) + response.raise_for_status() + payload = response.json() + series = payload.get("data", {}).get(tencent_symbol(symbol), {}) + klines = series.get("qfqday") or series.get("day") or [] + if not klines: + raise ValueError(f"empty Tencent ETF history for {symbol}") + rows.extend( + {"date": item[0], "symbol": normalize_symbol(symbol), "close": float(item[2])} + for item in klines + ) + break + except Exception: + if attempt == max(int(max_attempts), 1): + raise + time.sleep(max(float(retry_delay_seconds), 0.0) * attempt) + frame = pd.DataFrame(rows).drop_duplicates(["date", "symbol"], keep="last") + if frame.empty: + raise ValueError(f"empty Tencent ETF history for {symbol}") + return frame.sort_values("date").reset_index(drop=True) + + def fetch_yahoo_etf_history( symbol: str, *, @@ -160,9 +216,14 @@ def build_market_history_frame( frames: list[pd.DataFrame] = [] errors: dict[str, str] = {} requested_symbols = tuple(dict.fromkeys(normalize_symbol(symbol) for symbol in symbols)) - if source not in {"akshare", "yahoo"}: - raise ValueError("source must be 'akshare' or 'yahoo'") - fetcher = fetch_etf_history if source == "akshare" else fetch_yahoo_etf_history + fetchers = { + "akshare": fetch_etf_history, + "tencent": fetch_tencent_etf_history, + "yahoo": fetch_yahoo_etf_history, + } + if source not in fetchers: + raise ValueError("source must be 'akshare', 'tencent', or 'yahoo'") + fetcher = fetchers[source] for index, symbol in enumerate(requested_symbols): try: kwargs = {"start_date": start_date} @@ -208,7 +269,7 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--output", default="data/staging/market_history/etf_universe.latest.csv") parser.add_argument("--symbols", default=",".join(DEFAULT_ETF_SYMBOLS)) parser.add_argument("--start-date", default="20200101") - parser.add_argument("--source", choices=("akshare", "yahoo"), default="akshare") + parser.add_argument("--source", choices=("akshare", "tencent", "yahoo"), default="akshare") args = parser.parse_args(argv) symbols = tuple(symbol.strip() for symbol in args.symbols.split(",") if symbol.strip()) diagnostics = write_market_history_csv( diff --git a/tests/test_akshare_staging.py b/tests/test_akshare_staging.py index 62c3a6a..e8f2aa7 100644 --- a/tests/test_akshare_staging.py +++ b/tests/test_akshare_staging.py @@ -144,6 +144,8 @@ def test_build_market_history_frame_supports_yahoo_source(monkeypatch: pytest.Mo assert set(frame["symbol"]) == {"510300", "159915"} assert module.yahoo_symbol("510300") == "510300.SS" assert module.yahoo_symbol("159915") == "159915.SZ" + assert module.tencent_symbol("510300") == "sh510300" + assert module.tencent_symbol("159915") == "sz159915" def test_yahoo_history_preserves_adjusted_close_contract(monkeypatch: pytest.MonkeyPatch): diff --git a/tests/test_publish_market_history_workflow.py b/tests/test_publish_market_history_workflow.py index d43e3b7..ec54b4e 100644 --- a/tests/test_publish_market_history_workflow.py +++ b/tests/test_publish_market_history_workflow.py @@ -7,7 +7,7 @@ def test_publish_market_history_workflow_is_real_and_fail_closed() -> None: ).read_text(encoding="utf-8") assert "cneq-stage-akshare-market-history" in workflow - assert "--source yahoo" in workflow + assert "--source tencent" in workflow assert "--start-date 20220101" in workflow assert "cn_etf_market_history.csv" in workflow assert "actions/upload-artifact@v4" in workflow From 018c28cf2efa02e150d603722f2acdc08c786e8b Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 07:39:52 +0800 Subject: [PATCH 4/8] test: isolate optional HTTP dependency Co-Authored-By: Codex --- tests/test_akshare_staging.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/test_akshare_staging.py b/tests/test_akshare_staging.py index e8f2aa7..ad57a9c 100644 --- a/tests/test_akshare_staging.py +++ b/tests/test_akshare_staging.py @@ -1,6 +1,8 @@ from __future__ import annotations +import sys from pathlib import Path +from types import SimpleNamespace import pandas as pd import pytest @@ -170,9 +172,7 @@ def json(self) -> dict: } } - import requests - - monkeypatch.setattr(requests, "get", lambda *args, **kwargs: _Response()) + monkeypatch.setitem(sys.modules, "requests", SimpleNamespace(get=lambda *args, **kwargs: _Response())) frame = module.fetch_yahoo_etf_history("510300", start_date="20240101") From 9cfdf00943fa49cba313f0fff5b93ef4d35fdb2a Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 07:41:00 +0800 Subject: [PATCH 5/8] fix: make CN market-history rerun-safe Co-Authored-By: Codex --- .github/workflows/publish-market-history.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/publish-market-history.yml b/.github/workflows/publish-market-history.yml index c43323f..18f152e 100644 --- a/.github/workflows/publish-market-history.yml +++ b/.github/workflows/publish-market-history.yml @@ -44,7 +44,7 @@ jobs: - name: Upload market-history artifact uses: actions/upload-artifact@v4 with: - name: cn-equity-market-history-${{ github.run_id }} + name: cn-equity-market-history-${{ github.run_id }}-${{ github.run_attempt }} path: ${{ runner.temp }}/cn-market-history/cn_etf_market_history.csv if-no-files-found: error retention-days: 30 From 3465171dc3170ce676794c1608853d114026564a Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 07:57:56 +0800 Subject: [PATCH 6/8] fix: reject mixed ETF price bases Co-Authored-By: Codex --- .../akshare_market_history.py | 10 ++--- tests/test_akshare_staging.py | 44 +++++++++++++++++++ 2 files changed, 48 insertions(+), 6 deletions(-) diff --git a/src/cn_equity_snapshot_pipelines/akshare_market_history.py b/src/cn_equity_snapshot_pipelines/akshare_market_history.py index 58621ee..f83d8d5 100644 --- a/src/cn_equity_snapshot_pipelines/akshare_market_history.py +++ b/src/cn_equity_snapshot_pipelines/akshare_market_history.py @@ -91,7 +91,7 @@ def fetch_tencent_etf_history( response.raise_for_status() payload = response.json() series = payload.get("data", {}).get(tencent_symbol(symbol), {}) - klines = series.get("qfqday") or series.get("day") or [] + klines = series.get("qfqday") or [] if not klines: raise ValueError(f"empty Tencent ETF history for {symbol}") rows.extend( @@ -141,15 +141,13 @@ def fetch_yahoo_etf_history( series = result[0] timestamps = series.get("timestamp") or [] indicators = series.get("indicators") or {} - quotes = (indicators.get("quote") or [{}])[0] # Preserve the existing AkShare adjust="qfq" contract: `close` is adjusted, not raw exchange close. adjusted = (indicators.get("adjclose") or [{}])[0].get("adjclose") or [] - closes = quotes.get("close") or [] + if len(adjusted) != len(timestamps) or any(value is None for value in adjusted): + raise ValueError(f"incomplete Yahoo adjusted ETF history for {symbol}") rows = [] for index, raw_timestamp in enumerate(timestamps): - close = adjusted[index] if index < len(adjusted) else closes[index] if index < len(closes) else None - if close is None: - continue + close = adjusted[index] rows.append( { "date": pd.Timestamp.fromtimestamp(int(raw_timestamp), tz="UTC").date().isoformat(), diff --git a/tests/test_akshare_staging.py b/tests/test_akshare_staging.py index ad57a9c..7ae0787 100644 --- a/tests/test_akshare_staging.py +++ b/tests/test_akshare_staging.py @@ -178,3 +178,47 @@ def json(self) -> dict: assert frame.iloc[0]["close"] == 10.0 assert module.PRICE_BASIS == "adjusted_close" + + +def test_yahoo_history_rejects_incomplete_adjusted_series(monkeypatch: pytest.MonkeyPatch): + from cn_equity_snapshot_pipelines import akshare_market_history as module + + class _Response: + def raise_for_status(self) -> None: + return None + + def json(self) -> dict: + return { + "chart": { + "result": [ + { + "timestamp": [1704153600, 1704240000], + "indicators": { + "quote": [{"close": [12.0, 13.0]}], + "adjclose": [{"adjclose": [10.0]}], + }, + } + ] + } + } + + monkeypatch.setitem(sys.modules, "requests", SimpleNamespace(get=lambda *args, **kwargs: _Response())) + + with pytest.raises(ValueError, match="incomplete Yahoo adjusted ETF history"): + module.fetch_yahoo_etf_history("510300", start_date="20240101", max_attempts=1) + + +def test_tencent_history_rejects_raw_only_series(monkeypatch: pytest.MonkeyPatch): + from cn_equity_snapshot_pipelines import akshare_market_history as module + + class _Response: + def raise_for_status(self) -> None: + return None + + def json(self) -> dict: + return {"data": {"sh510300": {"day": [["2024-01-02", "10", "11"]]}}} + + monkeypatch.setitem(sys.modules, "requests", SimpleNamespace(get=lambda *args, **kwargs: _Response())) + + with pytest.raises(ValueError, match="empty Tencent ETF history"): + module.fetch_tencent_etf_history("510300", start_date="20240101", max_attempts=1) From 168812a07888d8a13714756a853d3ba39a10ea6c Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 08:11:08 +0800 Subject: [PATCH 7/8] fix: enforce complete ETF history contracts Co-Authored-By: Codex --- .../akshare_market_history.py | 72 ++++++++++++++++--- tests/test_akshare_staging.py | 33 +++++++-- 2 files changed, 90 insertions(+), 15 deletions(-) diff --git a/src/cn_equity_snapshot_pipelines/akshare_market_history.py b/src/cn_equity_snapshot_pipelines/akshare_market_history.py index f83d8d5..60f6d75 100644 --- a/src/cn_equity_snapshot_pipelines/akshare_market_history.py +++ b/src/cn_equity_snapshot_pipelines/akshare_market_history.py @@ -31,7 +31,9 @@ "512690", "159928", ) -PRICE_BASIS = "adjusted_close" +PRICE_BASIS = "adjusted_close_equivalent" +MAX_BOUNDARY_GAP_DAYS = 14 +MIN_BUSINESS_DAY_COVERAGE = 0.75 def normalize_symbol(value: object) -> str: @@ -59,17 +61,36 @@ def tencent_symbol(value: object) -> str: return f"{prefix}{symbol}" +def _validate_history_coverage( + frame: pd.DataFrame, + *, + symbol: str, + start_date: pd.Timestamp, + end_date: pd.Timestamp, +) -> None: + dates = pd.DatetimeIndex(pd.to_datetime(frame["date"], errors="coerce").dropna().unique()).sort_values() + expected = pd.bdate_range(start_date.normalize(), end_date.normalize()) + if ( + dates.empty + or dates.min() > start_date.normalize() + pd.Timedelta(days=MAX_BOUNDARY_GAP_DAYS) + or dates.max() < end_date.normalize() - pd.Timedelta(days=MAX_BOUNDARY_GAP_DAYS) + or len(dates) / max(len(expected), 1) < MIN_BUSINESS_DAY_COVERAGE + ): + raise ValueError(f"incomplete adjusted ETF history coverage for {symbol}") + + def fetch_tencent_etf_history( symbol: str, *, start_date: str = "20200101", + end_date: str | None = None, max_attempts: int = 3, retry_delay_seconds: float = 1.0, ) -> pd.DataFrame: import requests start = pd.Timestamp(start_date) - end = pd.Timestamp(datetime.now(timezone.utc).date()) + end = pd.Timestamp(end_date) if end_date else pd.Timestamp(datetime.now(timezone.utc).date()) rows: list[dict[str, object]] = [] for first_year in range(start.year, end.year + 1, 2): chunk_start = max(start, pd.Timestamp(first_year, 1, 1)) @@ -91,13 +112,28 @@ def fetch_tencent_etf_history( response.raise_for_status() payload = response.json() series = payload.get("data", {}).get(tencent_symbol(symbol), {}) - klines = series.get("qfqday") or [] + klines = series.get("qfqday") or series.get("day") or [] if not klines: raise ValueError(f"empty Tencent ETF history for {symbol}") - rows.extend( - {"date": item[0], "symbol": normalize_symbol(symbol), "close": float(item[2])} - for item in klines + basis = "tencent_qfq" if series.get("qfqday") else "tencent_qfq_identity" + chunk_frame = pd.DataFrame( + [ + { + "date": item[0], + "symbol": normalize_symbol(symbol), + "close": float(item[2]), + "price_basis": basis, + } + for item in klines + ] ) + _validate_history_coverage( + chunk_frame, + symbol=symbol, + start_date=chunk_start, + end_date=chunk_end, + ) + rows.extend(chunk_frame.to_dict("records")) break except Exception: if attempt == max(int(max_attempts), 1): @@ -113,13 +149,14 @@ def fetch_yahoo_etf_history( symbol: str, *, start_date: str = "20200101", + end_date: str | None = None, max_attempts: int = 3, retry_delay_seconds: float = 1.0, ) -> pd.DataFrame: import requests start = pd.Timestamp(start_date, tz="UTC") - end = pd.Timestamp(datetime.now(timezone.utc) + timedelta(days=1)) + end = pd.Timestamp(end_date, tz="UTC") if end_date else pd.Timestamp(datetime.now(timezone.utc) + timedelta(days=1)) query = urllib.parse.urlencode( { "period1": int(start.timestamp()), @@ -153,11 +190,18 @@ def fetch_yahoo_etf_history( "date": pd.Timestamp.fromtimestamp(int(raw_timestamp), tz="UTC").date().isoformat(), "symbol": normalize_symbol(symbol), "close": float(close), + "price_basis": "yahoo_adjusted_close", } ) frame = pd.DataFrame(rows) if frame.empty: raise ValueError(f"empty Yahoo ETF history for {symbol}") + _validate_history_coverage( + frame, + symbol=symbol, + start_date=start.tz_localize(None), + end_date=end.tz_localize(None), + ) return frame except Exception: if attempt == attempts: @@ -198,11 +242,19 @@ def fetch_etf_history( "date": pd.to_datetime(frame["日期"], errors="coerce").dt.date.astype(str), "symbol": normalize_symbol(symbol), "close": pd.to_numeric(frame["收盘"], errors="coerce"), + "price_basis": "akshare_qfq", } ) return output.dropna(subset=["date", "close"]) +def fetch_hybrid_etf_history(symbol: str, *, start_date: str = "20200101") -> pd.DataFrame: + try: + return fetch_yahoo_etf_history(symbol, start_date=start_date) + except Exception: + return fetch_tencent_etf_history(symbol, start_date=start_date) + + def build_market_history_frame( symbols: tuple[str, ...], *, @@ -216,11 +268,12 @@ def build_market_history_frame( requested_symbols = tuple(dict.fromkeys(normalize_symbol(symbol) for symbol in symbols)) fetchers = { "akshare": fetch_etf_history, + "hybrid": fetch_hybrid_etf_history, "tencent": fetch_tencent_etf_history, "yahoo": fetch_yahoo_etf_history, } if source not in fetchers: - raise ValueError("source must be 'akshare', 'tencent', or 'yahoo'") + raise ValueError("source must be 'akshare', 'hybrid', 'tencent', or 'yahoo'") fetcher = fetchers[source] for index, symbol in enumerate(requested_symbols): try: @@ -259,6 +312,7 @@ def write_market_history_csv( "start_date": start_date, "source": source, "price_basis": PRICE_BASIS, + "source_price_bases": sorted(frame["price_basis"].unique().tolist()), } @@ -267,7 +321,7 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--output", default="data/staging/market_history/etf_universe.latest.csv") parser.add_argument("--symbols", default=",".join(DEFAULT_ETF_SYMBOLS)) parser.add_argument("--start-date", default="20200101") - parser.add_argument("--source", choices=("akshare", "tencent", "yahoo"), default="akshare") + parser.add_argument("--source", choices=("akshare", "hybrid", "tencent", "yahoo"), default="akshare") args = parser.parse_args(argv) symbols = tuple(symbol.strip() for symbol in args.symbols.split(",") if symbol.strip()) diagnostics = write_market_history_csv( diff --git a/tests/test_akshare_staging.py b/tests/test_akshare_staging.py index 7ae0787..c0b7313 100644 --- a/tests/test_akshare_staging.py +++ b/tests/test_akshare_staging.py @@ -174,10 +174,10 @@ def json(self) -> dict: monkeypatch.setitem(sys.modules, "requests", SimpleNamespace(get=lambda *args, **kwargs: _Response())) - frame = module.fetch_yahoo_etf_history("510300", start_date="20240101") + frame = module.fetch_yahoo_etf_history("510300", start_date="20240102", end_date="20240102") assert frame.iloc[0]["close"] == 10.0 - assert module.PRICE_BASIS == "adjusted_close" + assert module.PRICE_BASIS == "adjusted_close_equivalent" def test_yahoo_history_rejects_incomplete_adjusted_series(monkeypatch: pytest.MonkeyPatch): @@ -205,10 +205,12 @@ def json(self) -> dict: monkeypatch.setitem(sys.modules, "requests", SimpleNamespace(get=lambda *args, **kwargs: _Response())) with pytest.raises(ValueError, match="incomplete Yahoo adjusted ETF history"): - module.fetch_yahoo_etf_history("510300", start_date="20240101", max_attempts=1) + module.fetch_yahoo_etf_history( + "510300", start_date="20240101", end_date="20240103", max_attempts=1 + ) -def test_tencent_history_rejects_raw_only_series(monkeypatch: pytest.MonkeyPatch): +def test_tencent_history_labels_identity_adjustment(monkeypatch: pytest.MonkeyPatch): from cn_equity_snapshot_pipelines import akshare_market_history as module class _Response: @@ -220,5 +222,24 @@ def json(self) -> dict: monkeypatch.setitem(sys.modules, "requests", SimpleNamespace(get=lambda *args, **kwargs: _Response())) - with pytest.raises(ValueError, match="empty Tencent ETF history"): - module.fetch_tencent_etf_history("510300", start_date="20240101", max_attempts=1) + frame = module.fetch_tencent_etf_history( + "510300", start_date="20240102", end_date="20240102", max_attempts=1 + ) + + assert set(frame["price_basis"]) == {"tencent_qfq_identity"} + + +def test_history_coverage_rejects_truncated_series() -> None: + from cn_equity_snapshot_pipelines import akshare_market_history as module + + frame = pd.DataFrame( + {"date": pd.bdate_range("2024-06-01", "2024-12-31"), "symbol": "510300", "close": 10.0} + ) + + with pytest.raises(ValueError, match="incomplete adjusted ETF history coverage"): + module._validate_history_coverage( + frame, + symbol="510300", + start_date=pd.Timestamp("2024-01-01"), + end_date=pd.Timestamp("2024-12-31"), + ) From af726b41cbba8045d497ad95afa69a25f4658325 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 11 Jul 2026 08:27:08 +0800 Subject: [PATCH 8/8] fix: honor Tencent listing inception and lock inputs Co-Authored-By: Codex --- .../akshare_market_history.py | 8 ++++- tests/test_akshare_staging.py | 31 +++++++++++++++++++ uv.lock | 2 ++ 3 files changed, 40 insertions(+), 1 deletion(-) diff --git a/src/cn_equity_snapshot_pipelines/akshare_market_history.py b/src/cn_equity_snapshot_pipelines/akshare_market_history.py index 60f6d75..e04f7fa 100644 --- a/src/cn_equity_snapshot_pipelines/akshare_market_history.py +++ b/src/cn_equity_snapshot_pipelines/akshare_market_history.py @@ -114,6 +114,8 @@ def fetch_tencent_etf_history( series = payload.get("data", {}).get(tencent_symbol(symbol), {}) klines = series.get("qfqday") or series.get("day") or [] if not klines: + if not rows: + break raise ValueError(f"empty Tencent ETF history for {symbol}") basis = "tencent_qfq" if series.get("qfqday") else "tencent_qfq_identity" chunk_frame = pd.DataFrame( @@ -130,7 +132,11 @@ def fetch_tencent_etf_history( _validate_history_coverage( chunk_frame, symbol=symbol, - start_date=chunk_start, + start_date=( + pd.Timestamp(chunk_frame["date"].min()) + if not rows + else chunk_start + ), end_date=chunk_end, ) rows.extend(chunk_frame.to_dict("records")) diff --git a/tests/test_akshare_staging.py b/tests/test_akshare_staging.py index c0b7313..c78d76b 100644 --- a/tests/test_akshare_staging.py +++ b/tests/test_akshare_staging.py @@ -229,6 +229,37 @@ def json(self) -> dict: assert set(frame["price_basis"]) == {"tencent_qfq_identity"} +def test_tencent_history_skips_pre_inception_chunks(monkeypatch: pytest.MonkeyPatch): + from cn_equity_snapshot_pipelines import akshare_market_history as module + + class _Response: + def __init__(self, params: dict[str, str]) -> None: + self.params = params + + def raise_for_status(self) -> None: + return None + + def json(self) -> dict: + if "2020-01-01" in self.params["param"]: + return {"data": {"sh510300": {"qfqday": []}}} + dates = pd.bdate_range("2022-06-01", "2023-12-29") + rows = [[date.date().isoformat(), "10", "11"] for date in dates] + return {"data": {"sh510300": {"qfqday": rows}}} + + monkeypatch.setitem( + sys.modules, + "requests", + SimpleNamespace(get=lambda *args, **kwargs: _Response(kwargs["params"])), + ) + + frame = module.fetch_tencent_etf_history( + "510300", start_date="20200101", end_date="20231231", max_attempts=1 + ) + + assert frame["date"].min() == "2022-06-01" + assert frame["date"].max() == "2023-12-29" + + def test_history_coverage_rejects_truncated_series() -> None: from cn_equity_snapshot_pipelines import akshare_market_history as module diff --git a/uv.lock b/uv.lock index 6720934..70144b4 100644 --- a/uv.lock +++ b/uv.lock @@ -242,6 +242,7 @@ dependencies = [ [package.optional-dependencies] public-data = [ { name = "akshare" }, + { name = "requests" }, ] test = [ { name = "pytest" }, @@ -253,6 +254,7 @@ requires-dist = [ { name = "cn-equity-strategies", git = "https://github.com/QuantStrategyLab/CnEquityStrategies.git?rev=73844e92a8570a61e5a9dc6c245809d0b27b89bc" }, { name = "pandas", specifier = ">=2.0" }, { name = "pytest", marker = "extra == 'test'", specifier = ">=8" }, + { name = "requests", marker = "extra == 'public-data'", specifier = ">=2.31" }, ] provides-extras = ["test", "public-data"]