diff options
Diffstat (limited to 'scripts')
| -rw-r--r-- | scripts/probe_akshare_nav_dates.py | 81 |
1 files changed, 81 insertions, 0 deletions
diff --git a/scripts/probe_akshare_nav_dates.py b/scripts/probe_akshare_nav_dates.py new file mode 100644 index 0000000..255cf91 --- /dev/null +++ b/scripts/probe_akshare_nav_dates.py @@ -0,0 +1,81 @@ +#!/usr/bin/env python3 +"""Runtime probe for real AKShare NAV dates. + +This is the "can the host actually reach AKShare?" check that backs the +`GET /api/market/nav-dates` endpoint. It talks to AKShare directly (no mocks), +memoises the akshare frame lookup with functools.lru_cache, and prints a single +JSON object on success: + + {"code": "000001", "dates": ["2026-09-21", "2026-09-18", ...]} + +On any failure (missing dependency, network error, unusable payload) it writes a +clear message to stderr and exits non-zero so callers can treat it as a skipped / +unavailable probe instead of inventing data. +""" + +import argparse +import functools +import json +import sys + +import akshare as ak + + [email protected]_cache(maxsize=8) +def fetch_nav_frame(code: str): + """Fetch the since-inception unit-NAV frame for a fund (cached per code).""" + return ak.fund_open_fund_info_em( + symbol=code, + indicator="单位净值走势", + period="成立来", + ) + + +def recent_nav_dates(code: str, limit: int = 5): + """Return the most recent distinct NAV dates (descending) for a fund code.""" + if len(code) != 6 or not code.isdigit(): + raise ValueError("fund code must contain exactly six digits") + + frame = fetch_nav_frame(code) + if frame is None or frame.empty: + raise RuntimeError(f"AKShare returned no NAV observations for {code}") + + column = "净值日期" + if column not in frame.columns: + raise RuntimeError(f"AKShare payload for {code} is missing the {column} column") + + dates = [] + for value in frame[column].tolist(): + text = str(value).strip()[:10] + if len(text) == 10 and text not in dates: + dates.append(text) + + dates.sort(reverse=True) + return dates[:limit] + + +def main(): + parser = argparse.ArgumentParser( + description="Probe real AKShare NAV dates for a fund code", + ) + parser.add_argument("--code", default="000001") + parser.add_argument("--limit", type=int, default=5) + args = parser.parse_args() + + try: + dates = recent_nav_dates(args.code, args.limit) + except Exception as error: # noqa: BLE001 - probe reports the concrete reason + print(f"AKShare NAV-date probe failed for {args.code}: {error}", file=sys.stderr) + return 2 + + if not dates: + print(f"AKShare NAV-date probe found no dates for {args.code}", file=sys.stderr) + return 2 + + json.dump({"code": args.code, "dates": dates}, sys.stdout, ensure_ascii=False) + sys.stdout.write("\n") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) |