diff --git a/.github/workflows/publish-market-history.yml b/.github/workflows/publish-market-history.yml index bb67404..18f152e 100644 --- a/.github/workflows/publish-market-history.yml +++ b/.github/workflows/publish-market-history.yml @@ -31,19 +31,20 @@ 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 tencent \ --start-date 20220101 \ --output "${RUNNER_TEMP}/cn-market-history/cn_etf_market_history.csv" - 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 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..e04f7fa 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 @@ -30,6 +31,9 @@ "512690", "159928", ) +PRICE_BASIS = "adjusted_close_equivalent" +MAX_BOUNDARY_GAP_DAYS = 14 +MIN_BUSINESS_DAY_COVERAGE = 0.75 def normalize_symbol(value: object) -> str: @@ -45,6 +49,173 @@ 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 tencent_symbol(value: object) -> str: + symbol = normalize_symbol(value) + prefix = "sz" if symbol.startswith(("0", "1", "3")) else "sh" + 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(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)) + 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: + 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( + [ + { + "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=( + pd.Timestamp(chunk_frame["date"].min()) + if not rows + else chunk_start + ), + end_date=chunk_end, + ) + rows.extend(chunk_frame.to_dict("records")) + 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, + *, + 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(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()), + "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 {} + # Preserve the existing AkShare adjust="qfq" contract: `close` is adjusted, not raw exchange close. + adjusted = (indicators.get("adjclose") or [{}])[0].get("adjclose") 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] + rows.append( + { + "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: + raise + time.sleep(max(float(retry_delay_seconds), 0.0) * attempt) + raise AssertionError("unreachable") + + def fetch_etf_history( symbol: str, *, @@ -77,24 +248,45 @@ 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, ...], *, 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)) + 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', 'hybrid', 'tencent', or 'yahoo'") + fetcher = fetchers[source] 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 +304,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 +316,9 @@ def write_market_history_csv( "row_count": int(len(frame)), "symbols": sorted(frame["symbol"].unique().tolist()), "start_date": start_date, + "source": source, + "price_basis": PRICE_BASIS, + "source_price_bases": sorted(frame["price_basis"].unique().tolist()), } @@ -131,12 +327,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", "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( 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..c78d76b 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 @@ -122,3 +124,153 @@ 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" + assert module.tencent_symbol("510300") == "sh510300" + assert module.tencent_symbol("159915") == "sz159915" + + +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]}], + }, + } + ] + } + } + + monkeypatch.setitem(sys.modules, "requests", SimpleNamespace(get=lambda *args, **kwargs: _Response())) + + 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_equivalent" + + +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", end_date="20240103", max_attempts=1 + ) + + +def test_tencent_history_labels_identity_adjustment(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())) + + 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_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 + + 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"), + ) diff --git a/tests/test_publish_market_history_workflow.py b/tests/test_publish_market_history_workflow.py index e9f1b99..ec54b4e 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 tencent" in workflow assert "--start-date 20220101" in workflow assert "cn_etf_market_history.csv" in workflow assert "actions/upload-artifact@v4" in workflow 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"]