1#!/usr/bin/env python3
2"""
3A股股票筛选器
4根据多种财务指标筛选符合条件的股票
5 
6依赖: pip install akshare pandas numpy
7"""
8 
9import argparse
10import json
11import sys
12import time
13from datetime import datetime
14from typing import List, Dict
15from functools import wraps
16 
17try:
18    import akshare as ak
19    import pandas as pd
20    import numpy as np
21except ImportError:
22    print("错误: 请先安装依赖库")
23    print("pip install akshare pandas numpy")
24    sys.exit(1)
25 
26 
27def retry_on_failure(max_retries: int = 3, delay: float = 1.0):
28    """网络请求重试装饰器"""
29    def decorator(func):
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                        print(f"  重试 ({attempt + 1}/{max_retries})...")
40                        time.sleep(delay * (attempt + 1))
41            raise last_error
42        return wrapper
43    return decorator
44 
45 
46INDEX_CODE_MAP = {
47    "hs300": "000300",
48    "zz500": "000905",
49    "zz1000": "000852",
50    "cyb": "399006",
51    "kcb": "000688"
52}
53 
54 
55class StockScreener:
56    """股票筛选器"""
57 
58    def __init__(self):
59        self.all_stocks_data = None
60 
61    def load_stock_data(self, scope: str = "hs300", custom_codes: List[str] = None) -> pd.DataFrame:
62        """加载股票数据"""
63        print(f"正在加载股票数据 (范围: {scope})...")
64 
65        try:
66            if scope == "all":
67                df = ak.stock_zh_a_spot_em()
68            elif scope in INDEX_CODE_MAP:
69                df = self._get_index_stocks_data(INDEX_CODE_MAP[scope])
70            elif scope.startswith("custom:") or custom_codes:
71                codes = custom_codes or scope.replace("custom:", "").split(",")
72                df = self._get_custom_stocks_data(codes)
73            else:
74                df = ak.stock_zh_a_spot_em()
75 
76            self.all_stocks_data = df
77            print(f"已加载 {len(df)} 只股票数据")
78            return df
79 
80        except Exception as e:
81            print(f"加载数据失败: {e}")
82            return pd.DataFrame()
83 
84    @retry_on_failure(max_retries=3, delay=2.0)
85    def _get_all_stocks_realtime(self) -> pd.DataFrame:
86        """获取全部A股实时数据（带重试）"""
87        return ak.stock_zh_a_spot_em()
88 
89    @retry_on_failure(max_retries=3, delay=2.0)
90    def _get_index_constituents(self, index_code: str) -> list:
91        """获取指数成分股列表（带重试）"""
92        df = ak.index_stock_cons(symbol=index_code)
93        return df['品种代码'].tolist()
94 
95    def _get_index_stocks_data(self, index_code: str) -> pd.DataFrame:
96        """获取指数成分股数据"""
97        try:
98            # 获取成分股列表
99            print(f"  获取指数 {index_code} 成分股...")
100            codes = self._get_index_constituents(index_code)
101            print(f"  成分股数量: {len(codes)}")
102 
103            # 获取实时数据
104            print("  获取实时行情...")
105            all_stocks = self._get_all_stocks_realtime()
106            df = all_stocks[all_stocks['代码'].isin(codes)]
107            return df
108        except Exception as e:
109            print(f"获取指数成分股失败: {e}")
110            return pd.DataFrame()
111 
112    def _get_custom_stocks_data(self, codes: List[str]) -> pd.DataFrame:
113        """获取自定义股票列表数据"""
114        try:
115            all_stocks = self._get_all_stocks_realtime()
116            df = all_stocks[all_stocks['代码'].isin(codes)]
117            return df
118        except Exception as e:
119            print(f"获取自定义股票数据失败: {e}")
120            return pd.DataFrame()
121 
122    def _apply_numeric_filter(self, df: pd.DataFrame, column: str,
123                               min_val: float = None, max_val: float = None) -> pd.DataFrame:
124        """应用数值筛选条件"""
125        if column not in df.columns:
126            return df
127 
128        numeric_col = pd.to_numeric(df[column], errors='coerce')
129        if min_val is not None:
130            df = df[numeric_col >= min_val]
131        if max_val is not None:
132            df = df[numeric_col <= max_val]
133        return df
134 
135    def _find_column(self, df: pd.DataFrame, candidates: List[str]) -> str:
136        """从候选列名中找到存在的列"""
137        for col in candidates:
138            if col in df.columns:
139                return col
140        return None
141 
142    def apply_filters(self, df: pd.DataFrame, filters: Dict) -> pd.DataFrame:
143        """应用筛选条件"""
144        filtered = df.copy()
145 
146        # PE筛选
147        filtered = self._apply_numeric_filter(
148            filtered, '市盈率-动态',
149            min_val=filters.get('pe_min'),
150            max_val=filters.get('pe_max')
151        )
152 
153        # PB筛选
154        filtered = self._apply_numeric_filter(
155            filtered, '市净率',
156            min_val=filters.get('pb_min'),
157            max_val=filters.get('pb_max')
158        )
159 
160        # ROE筛选
161        if filters.get('roe_min') is not None:
162            roe_col = self._find_column(filtered, ['净资产收益率', 'ROE', '加权净资产收益率'])
163            if roe_col:
164                filtered = self._apply_numeric_filter(filtered, roe_col, min_val=filters['roe_min'])
165 
166        # 资产负债率筛选
167        filtered = self._apply_numeric_filter(
168            filtered, '资产负债率',
169            max_val=filters.get('debt_ratio_max')
170        )
171 
172        # 总市值筛选（转换为亿）
173        if '总市值' in filtered.columns:
174            if filters.get('market_cap_min') is not None or filters.get('market_cap_max') is not None:
175                filtered['总市值_亿'] = pd.to_numeric(filtered['总市值'], errors='coerce') / 1e8
176                filtered = self._apply_numeric_filter(
177                    filtered, '总市值_亿',
178                    min_val=filters.get('market_cap_min'),
179                    max_val=filters.get('market_cap_max')
180                )
181 
182        return filtered
183 
184    def _get_numeric_value(self, row: pd.Series, column: str) -> float:
185        """从行中获取数值，无效返回 NaN"""
186        return pd.to_numeric(row.get(column, np.nan), errors='coerce')
187 
188    def calculate_score(self, row: pd.Series) -> float:
189        """计算综合评分 (0-100)"""
190        score = 50
191 
192        try:
193            # PE评分 (越低越好, 负数除外)
194            pe = self._get_numeric_value(row, '市盈率-动态')
195            if not np.isnan(pe) and pe > 0:
196                if pe < 10:
197                    score += 15
198                elif pe < 15:
199                    score += 10
200                elif pe < 20:
201                    score += 5
202                elif pe > 50:
203                    score -= 10
204 
205            # PB评分
206            pb = self._get_numeric_value(row, '市净率')
207            if not np.isnan(pb) and pb > 0:
208                if 0.5 < pb < 1.5:
209                    score += 10
210                elif 1.5 <= pb < 3:
211                    score += 5
212                elif pb > 5:
213                    score -= 5
214 
215            # ROE评分
216            roe_col = self._find_column(row.index.to_frame(), ['净资产收益率', 'ROE', '加权净资产收益率'])
217            if roe_col:
218                roe = self._get_numeric_value(row, roe_col)
219                if not np.isnan(roe):
220                    if roe > 20:
221                        score += 15
222                    elif roe > 15:
223                        score += 10
224                    elif roe > 10:
225                        score += 5
226                    elif roe < 5:
227                        score -= 5
228 
229            # 涨跌幅评分 (下跌可能是机会)
230            change = self._get_numeric_value(row, '涨跌幅')
231            if not np.isnan(change):
232                if -5 < change < 0:
233                    score += 3
234                elif change < -5:
235                    score += 5
236 
237        except Exception:
238            pass
239 
240        return max(0, min(100, score))
241 
242    def screen(self, scope: str = "hs300", filters: Dict = None,
243              sort_by: str = "score", top_n: int = None) -> List[Dict]:
244        """执行筛选"""
245        # 加载数据
246        if scope.startswith("custom:"):
247            codes = scope.replace("custom:", "").split(",")
248            df = self.load_stock_data(scope="custom", custom_codes=codes)
249        else:
250            df = self.load_stock_data(scope=scope)
251 
252        if df.empty:
253            return []
254 
255        # 应用筛选条件
256        if filters:
257            df = self.apply_filters(df, filters)
258 
259        if df.empty:
260            return []
261 
262        # 计算评分
263        df['评分'] = df.apply(self.calculate_score, axis=1)
264 
265        # 排序
266        if sort_by == "score":
267            df = df.sort_values('评分', ascending=False)
268        elif sort_by == "pe":
269            pe_col = '市盈率-动态' if '市盈率-动态' in df.columns else None
270            if pe_col:
271                df = df.sort_values(pe_col, ascending=True)
272        elif sort_by == "pb":
273            if '市净率' in df.columns:
274                df = df.sort_values('市净率', ascending=True)
275        elif sort_by == "market_cap":
276            if '总市值' in df.columns:
277                df = df.sort_values('总市值', ascending=False)
278 
279        # 限制数量
280        if top_n:
281            df = df.head(top_n)
282 
283        # 转换为结果列表
284        results = []
285        for _, row in df.iterrows():
286            result = {
287                "代码": row.get('代码', ''),
288                "名称": row.get('名称', ''),
289                "最新价": row.get('最新价', ''),
290                "涨跌幅": row.get('涨跌幅', ''),
291                "市盈率": row.get('市盈率-动态', ''),
292                "市净率": row.get('市净率', ''),
293                "总市值(亿)": round(float(row.get('总市值', 0)) / 100000000, 2) if row.get('总市值') else '',
294                "评分": row.get('评分', 50)
295            }
296            results.append(result)
297 
298        return results
299 
300 
301def main():
302    parser = argparse.ArgumentParser(description="A股股票筛选器")
303    parser.add_argument("--scope", type=str, default="hs300",
304                       help="筛选范围: all/hs300/zz500/zz1000/cyb/kcb/custom:代码1,代码2")
305    parser.add_argument("--pe-max", type=float, help="最大PE")
306    parser.add_argument("--pe-min", type=float, help="最小PE")
307    parser.add_argument("--pb-max", type=float, help="最大PB")
308    parser.add_argument("--pb-min", type=float, help="最小PB")
309    parser.add_argument("--roe-min", type=float, help="最小ROE (%)")
310    parser.add_argument("--debt-ratio-max", type=float, help="最大资产负债率 (%)")
311    parser.add_argument("--dividend-min", type=float, help="最小股息率 (%)")
312    parser.add_argument("--market-cap-min", type=float, help="最小市值 (亿)")
313    parser.add_argument("--market-cap-max", type=float, help="最大市值 (亿)")
314    parser.add_argument("--sort-by", type=str, default="score",
315                       choices=["score", "pe", "pb", "market_cap"],
316                       help="排序方式")
317    parser.add_argument("--top", type=int, default=50, help="返回前N只股票")
318    parser.add_argument("--output", type=str, help="输出文件路径 (JSON)")
319 
320    args = parser.parse_args()
321 
322    # 构建筛选条件
323    filter_keys = [
324        'pe_max', 'pe_min', 'pb_max', 'pb_min', 'roe_min',
325        'debt_ratio_max', 'dividend_min', 'market_cap_min', 'market_cap_max'
326    ]
327    filters = {
328        k: getattr(args, k.replace('-', '_'))
329        for k in filter_keys
330        if getattr(args, k.replace('-', '_')) is not None
331    }
332 
333    # 执行筛选
334    screener = StockScreener()
335    results = screener.screen(
336        scope=args.scope,
337        filters=filters if filters else None,
338        sort_by=args.sort_by,
339        top_n=args.top
340    )
341 
342    # 输出结果
343    output = {
344        "screen_time": datetime.now().isoformat(),
345        "scope": args.scope,
346        "filters": filters,
347        "count": len(results),
348        "results": results
349    }
350 
351    output_json = json.dumps(output, ensure_ascii=False, indent=2, default=str)
352 
353    if args.output:
354        with open(args.output, 'w', encoding='utf-8') as f:
355            f.write(output_json)
356        print(f"筛选结果已保存到: {args.output}")
357        print(f"共筛选出 {len(results)} 只股票")
358    else:
359        print(output_json)
360 
361 
362if __name__ == "__main__":
363    main()
364