1#!/usr/bin/env python3
2"""
3A股数据获取模块
4使用akshare获取股票财务数据、行情数据、股东信息等
5 
6依赖: pip install akshare pandas
7"""
8 
9import argparse
10import json
11import sys
12import time
13import os
14from datetime import datetime, timedelta
15from typing import Optional, Callable
16from functools import wraps
17 
18try:
19    import akshare as ak
20    import pandas as pd
21except ImportError:
22    print("错误: 请先安装依赖库")
23    print("pip install akshare pandas")
24    sys.exit(1)
25 
26 
27def retry_on_failure(max_retries: int = 3, delay: float = 1.0):
28    """网络请求重试装饰器"""
29    def decorator(func: Callable):
30        @wraps(func)
31        def wrapper(*args, **kwargs):
32            last_error = None
33            for attempt in range(max_retries):
34                try:
35                    return func(*args, **kwargs)
36                except Exception as e:
37                    last_error = e
38                    if attempt < max_retries - 1:
39                        time.sleep(delay * (attempt + 1))  # 递增等待
40            return {"error": f"重试{max_retries}次后失败: {str(last_error)}"}
41        return wrapper
42    return decorator
43 
44 
45def safe_float(value) -> Optional[float]:
46    """安全转换为浮点数"""
47    if value is None or value == '' or value == '--':
48        return None
49    try:
50        if pd.isna(value):
51            return None
52        if isinstance(value, str):
53            value = value.replace('%', '').replace(',', '').replace('亿', '')
54        return float(value)
55    except (ValueError, TypeError):
56        return None
57 
58 
59def get_cache_path(code: str, data_type: str) -> str:
60    """获取缓存文件路径"""
61    cache_dir = os.path.join(os.path.dirname(__file__), '.cache')
62    os.makedirs(cache_dir, exist_ok=True)
63    today = datetime.now().strftime('%Y%m%d')
64    return os.path.join(cache_dir, f"{code}_{data_type}_{today}.json")
65 
66 
67def load_cache(code: str, data_type: str) -> Optional[dict]:
68    """加载缓存数据（当天有效）"""
69    cache_path = get_cache_path(code, data_type)
70    if os.path.exists(cache_path):
71        try:
72            with open(cache_path, 'r', encoding='utf-8') as f:
73                return json.load(f)
74        except (json.JSONDecodeError, IOError):
75            return None
76    return None
77 
78 
79def save_cache(code: str, data_type: str, data: dict):
80    """保存缓存数据"""
81    cache_path = get_cache_path(code, data_type)
82    try:
83        with open(cache_path, 'w', encoding='utf-8') as f:
84            json.dump(data, f, ensure_ascii=False, default=str)
85    except IOError:
86        pass
87 
88 
89@retry_on_failure(max_retries=2, delay=1.0)
90def get_stock_info(code: str) -> dict:
91    """获取股票基本信息"""
92    try:
93        df = ak.stock_individual_info_em(symbol=code)
94        info = {}
95        for _, row in df.iterrows():
96            info[row['item']] = row['value']
97        return {
98            "code": code,
99            "name": info.get("股票简称", ""),
100            "industry": info.get("行业", ""),
101            "market_cap": safe_float(info.get("总市值")),
102            "float_cap": safe_float(info.get("流通市值")),
103            "total_shares": safe_float(info.get("总股本")),
104            "float_shares": safe_float(info.get("流通股")),
105            "pe_ttm": safe_float(info.get("市盈率(动态)")),
106            "pb": safe_float(info.get("市净率")),
107            "listing_date": info.get("上市时间", "")
108        }
109    except Exception as e:
110        return {"code": code, "error": str(e)}
111 
112 
113@retry_on_failure(max_retries=2, delay=1.0)
114def get_financial_data(code: str, years: int = 3) -> dict:
115    """获取财务数据（资产负债表、利润表、现金流量表）"""
116    max_records = min(years * 4, 12)
117    result = {
118        "balance_sheet": [],
119        "income_statement": [],
120        "cash_flow": []
121    }
122 
123    fetch_configs = [
124        ("balance_sheet", ak.stock_balance_sheet_by_report_em),
125        ("income_statement", ak.stock_profit_sheet_by_report_em),
126        ("cash_flow", ak.stock_cash_flow_sheet_by_report_em),
127    ]
128 
129    for key, fetch_func in fetch_configs:
130        try:
131            df = fetch_func(symbol=code)
132            if df is not None and not df.empty:
133                result[key] = df.head(max_records).to_dict(orient='records')
134        except Exception as e:
135            result[f"{key}_error"] = str(e)
136 
137    return result
138 
139 
140def get_financial_indicators(code: str, limit: int = 8) -> dict:
141    """获取财务指标，优先使用快速API，失败时降级到备用API"""
142    apis = [ak.stock_financial_abstract, ak.stock_financial_analysis_indicator]
143 
144    for api in apis:
145        try:
146            df = api(symbol=code)
147            if df is not None and not df.empty:
148                return df.head(limit).to_dict(orient='records')
149        except Exception:
150            continue
151 
152    return []
153 
154 
155def get_valuation_data(code: str) -> dict:
156    """获取估值数据"""
157    result = {}
158 
159    try:
160        df = ak.stock_a_ttm_lyr(symbol=code)
161        if df is None or df.empty:
162            return result
163 
164        latest = df.iloc[-1].to_dict()
165        result["latest"] = latest
166        result["history_count"] = len(df)
167 
168        for col in ['pe_ttm', 'pb']:
169            val = latest.get(col)
170            if val and not pd.isna(val):
171                result[f"{col}_percentile"] = (df[col].dropna() < val).mean() * 100
172 
173    except Exception as e:
174        result["error"] = str(e)
175        result["note"] = "估值历史数据获取失败，将使用基本信息中的估值"
176 
177    return result
178 
179 
180@retry_on_failure(max_retries=2, delay=1.0)
181def get_holder_data(code: str) -> dict:
182    """获取股东信息"""
183    result = {}
184 
185    try:
186        df_top10 = ak.stock_gdfx_top_10_em(symbol=code)
187        if df_top10 is not None and not df_top10.empty:
188            result["top_10_holders"] = df_top10.head(10).to_dict(orient='records')
189    except Exception as e:
190        result["top_10_holders_error"] = str(e)
191 
192    try:
193        df_holder_num = ak.stock_zh_a_gdhs(symbol=code)
194        if df_holder_num is not None and not df_holder_num.empty:
195            result["holder_count_history"] = df_holder_num.head(10).to_dict(orient='records')
196    except Exception as e:
197        result["holder_count_error"] = str(e)
198 
199    return result
200 
201 
202@retry_on_failure(max_retries=2, delay=1.0)
203def get_dividend_data(code: str) -> dict:
204    """获取分红数据，优先使用主API，失败时降级到备用API"""
205    apis = [
206        lambda c: ak.stock_dividend_cninfo(symbol=c),
207        lambda c: ak.stock_history_dividend_detail(symbol=c, indicator="分红"),
208    ]
209 
210    for api in apis:
211        try:
212            df = api(code)
213            if df is not None and not df.empty:
214                return {
215                    "dividend_history": df.to_dict(orient='records'),
216                    "dividend_count": len(df)
217                }
218        except Exception:
219            continue
220 
221    return {"dividend_history": [], "dividend_count": 0}
222 
223 
224@retry_on_failure(max_retries=2, delay=1.0)
225def get_price_data(code: str, days: int = 60) -> dict:
226    """获取价格数据"""
227    try:
228        end_date = datetime.now().strftime('%Y%m%d')
229        start_date = (datetime.now() - timedelta(days=days)).strftime('%Y%m%d')
230 
231        df = ak.stock_zh_a_hist(symbol=code, period="daily",
232                                start_date=start_date, end_date=end_date, adjust="qfq")
233        if df is not None and not df.empty:
234            latest = df.iloc[-1]
235            return {
236                "latest_price": safe_float(latest['收盘']),
237                "latest_date": str(latest['日期']),
238                "price_change_pct": safe_float(latest['涨跌幅']),
239                "volume": safe_float(latest['成交量']),
240                "turnover": safe_float(latest['成交额']),
241                "high_60d": safe_float(df['最高'].max()),
242                "low_60d": safe_float(df['最低'].min()),
243                "avg_volume_20d": safe_float(df.tail(20)['成交量'].mean()),
244                "price_data": df.tail(30).to_dict(orient='records')  # 只保留30天
245            }
246        return {}
247    except Exception as e:
248        return {"error": str(e)}
249 
250 
251@retry_on_failure(max_retries=2, delay=1.0)
252def get_index_constituents(index_name: str) -> list:
253    """获取指数成分股"""
254    index_map = {
255        "hs300": "000300",
256        "zz500": "000905",
257        "zz1000": "000852",
258        "cyb": "399006",
259        "kcb": "000688"
260    }
261 
262    index_code = index_map.get(index_name)
263    if not index_code:
264        return []
265 
266    try:
267        df = ak.index_stock_cons(symbol=index_code)
268        if df is not None and not df.empty:
269            return df['品种代码'].tolist()
270        return []
271    except Exception as e:
272        print(f"获取指数成分股失败: {e}")
273        return []
274 
275 
276def get_all_a_stocks() -> list:
277    """获取全部A股代码"""
278    try:
279        df = ak.stock_zh_a_spot_em()
280        if df is not None and not df.empty:
281            return df['代码'].tolist()
282        return []
283    except Exception as e:
284        print(f"获取全部A股失败: {e}")
285        return []
286 
287 
288def fetch_stock_data(code: str, data_type: str = "all", years: int = 3, use_cache: bool = True) -> dict:
289    """获取单只股票的数据"""
290    # 尝试加载缓存
291    if use_cache:
292        cached = load_cache(code, data_type)
293        if cached:
294            print(f"使用缓存数据: {code}")
295            return cached
296 
297    result = {
298        "code": code,
299        "fetch_time": datetime.now().isoformat(),
300        "data_type": data_type
301    }
302 
303    print(f"正在获取 {code} 的数据...")
304 
305    if data_type in ["all", "basic"]:
306        print("  - 获取基本信息...")
307        result["basic_info"] = get_stock_info(code)
308 
309    if data_type in ["all", "financial"]:
310        print("  - 获取财务数据...")
311        result["financial_data"] = get_financial_data(code, years)
312        print("  - 获取财务指标...")
313        result["financial_indicators"] = get_financial_indicators(code)
314 
315    if data_type in ["all", "valuation"]:
316        print("  - 获取估值数据...")
317        result["valuation"] = get_valuation_data(code)
318        print("  - 获取价格数据...")
319        result["price"] = get_price_data(code)
320 
321    if data_type in ["all", "holder"]:
322        print("  - 获取股东数据...")
323        result["holder"] = get_holder_data(code)
324        print("  - 获取分红数据...")
325        result["dividend"] = get_dividend_data(code)
326 
327    # 保存缓存
328    if use_cache:
329        save_cache(code, data_type, result)
330 
331    print(f"数据获取完成: {code}")
332    return result
333 
334 
335def fetch_multiple_stocks(codes: list, data_type: str = "basic") -> dict:
336    """获取多只股票数据"""
337    result = {
338        "fetch_time": datetime.now().isoformat(),
339        "stocks": [],
340        "success_count": 0,
341        "fail_count": 0
342    }
343 
344    total = len(codes)
345    for i, code in enumerate(codes):
346        print(f"[{i+1}/{total}] 获取 {code}...")
347        try:
348            stock_data = fetch_stock_data(code, data_type, use_cache=True)
349            if "error" not in stock_data.get("basic_info", {}):
350                result["stocks"].append(stock_data)
351                result["success_count"] += 1
352            else:
353                result["fail_count"] += 1
354        except Exception as e:
355            print(f"  获取失败: {e}")
356            result["fail_count"] += 1
357 
358        # 避免请求过快
359        if i < total - 1:
360            time.sleep(0.5)
361 
362    return result
363 
364 
365def main():
366    parser = argparse.ArgumentParser(description="A股数据获取工具")
367    parser.add_argument("--code", type=str, help="股票代码 (如: 600519)")
368    parser.add_argument("--codes", type=str, help="多个股票代码，逗号分隔 (如: 600519,000858)")
369    parser.add_argument("--data-type", type=str, default="basic",
370                       choices=["all", "basic", "financial", "valuation", "holder"],
371                       help="数据类型 (默认: basic)")
372    parser.add_argument("--years", type=int, default=3, help="获取多少年的历史数据 (默认: 3)")
373    parser.add_argument("--scope", type=str, help="筛选范围: hs300/zz500/cyb/kcb/all")
374    parser.add_argument("--no-cache", action="store_true", help="不使用缓存")
375    parser.add_argument("--output", type=str, help="输出文件路径 (JSON)")
376 
377    args = parser.parse_args()
378 
379    result = {}
380 
381    if args.code:
382        result = fetch_stock_data(args.code, args.data_type, args.years,
383                                   use_cache=not args.no_cache)
384    elif args.codes:
385        codes = [c.strip() for c in args.codes.split(",")]
386        result = fetch_multiple_stocks(codes, args.data_type)
387    elif args.scope:
388        if args.scope == "all":
389            codes = get_all_a_stocks()
390        else:
391            codes = get_index_constituents(args.scope)
392        result = {"scope": args.scope, "stocks": codes, "count": len(codes)}
393    else:
394        print("请提供 --code, --codes 或 --scope 参数")
395        sys.exit(1)
396 
397    # 输出结果
398    output = json.dumps(result, ensure_ascii=False, indent=2, default=str)
399 
400    if args.output:
401        with open(args.output, 'w', encoding='utf-8') as f:
402            f.write(output)
403        print(f"\n数据已保存到: {args.output}")
404    else:
405        print(output)
406 
407 
408if __name__ == "__main__":
409    main()
410