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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
|
#!/usr/bin/env python3
"""Small raw-data adapter for the AKShare endpoints used by Fund Lab."""
import argparse
import datetime as dt
import json
import re
import sys
from decimal import Decimal, InvalidOperation
import akshare as ak
import pandas as pd
SCHEMA_VERSION = "fund-lab.akshare.v1"
def collected_at():
return dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z")
def source_revision():
return f"akshare-{getattr(ak, '__version__', 'unknown')}/eastmoney"
def is_missing(value):
try:
return bool(pd.isna(value))
except (TypeError, ValueError):
return False
def text(value):
if is_missing(value):
return None
value = str(value).strip()
return value or None
def fund_code(value):
value = text(value)
if value is None:
return None
if value.isdigit() and len(value) < 6:
value = value.zfill(6)
return value if re.fullmatch(r"\d{6}", value) else None
def decimal_text(value):
value = text(value)
if value is None:
return None
try:
return format(Decimal(value), "f")
except InvalidOperation:
return None
def date_text(value):
if is_missing(value):
return None
if isinstance(value, (dt.datetime, dt.date)):
return value.strftime("%Y-%m-%d")
parsed = pd.to_datetime(value, errors="coerce")
if pd.isna(parsed):
return None
return parsed.strftime("%Y-%m-%d")
def search(query):
query = query.strip()
if not query:
raise ValueError("search query cannot be empty")
frame = ak.fund_name_em()
query_lower = query.casefold()
rows = []
for _, row in frame.iterrows():
code = fund_code(row.get("基金代码"))
name = text(row.get("基金简称"))
pinyin = text(row.get("拼音缩写"))
full_pinyin = text(row.get("拼音全称"))
if code is None or name is None:
continue
searchable = [code, name, pinyin or "", full_pinyin or ""]
if not any(query_lower in value.casefold() for value in searchable):
continue
rows.append(
{
"code": code,
"name": name,
"fund_type": text(row.get("基金类型")),
"rank": 0 if code == query else (1 if name == query else 2),
}
)
rows.sort(key=lambda item: (item.pop("rank"), item["code"]))
return {
"schema_version": SCHEMA_VERSION,
"operation": "search",
"source": "akshare",
"source_revision": source_revision(),
"collected_at": collected_at(),
"instruments": rows[:100],
}
def nav(code):
if not re.fullmatch(r"\d{6}", code):
raise ValueError("fund code must contain exactly six digits")
frame = ak.fund_open_fund_info_em(
symbol=code,
indicator="单位净值走势",
period="成立来",
)
observations = []
for _, row in frame.iterrows():
nav_date = date_text(row.get("净值日期"))
nav_value = decimal_text(row.get("单位净值"))
if nav_date is None or nav_value is None:
continue
observations.append(
{
"nav_date": nav_date,
"published_at": None,
"nav": nav_value,
"accumulated_nav": None,
"daily_return": decimal_text(row.get("日增长率")),
}
)
if not observations:
raise ValueError(f"AKShare returned no usable NAV observations for {code}")
return {
"schema_version": SCHEMA_VERSION,
"operation": "nav",
"source": "akshare",
"source_revision": source_revision(),
"collected_at": collected_at(),
"instrument": {"code": code},
"observations": observations,
}
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--operation", choices=("search", "nav"), required=True)
parser.add_argument("--query")
parser.add_argument("--code")
args = parser.parse_args()
try:
payload = search(args.query) if args.operation == "search" else nav(args.code)
json.dump(payload, sys.stdout, ensure_ascii=False, separators=(",", ":"))
sys.stdout.write("\n")
return 0
except Exception as error:
print(f"AKShare collector failed: {error}", file=sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main())
|
