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
|
#!/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
@functools.lru_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())
|
