1#!/usr/bin/env python3
2"""
3A股财务分析模块
4提供财务健康度分析、杜邦分析、异常检测等功能
5 
6依赖: pip install akshare pandas numpy
7"""
8 
9import argparse
10import json
11import sys
12from datetime import datetime
13from typing import Optional, List, Dict
14 
15try:
16    import pandas as pd
17    import numpy as np
18except ImportError:
19    print("错误: 请先安装依赖库")
20    print("pip install pandas numpy")
21    sys.exit(1)
22 
23 
24class FinancialAnalyzer:
25    """财务分析器"""
26 
27    def __init__(self, stock_data: Dict = None):
28        self.stock_data = stock_data or {}
29        self.analysis_result = {}
30 
31    def load_data(self, file_path: str):
32        """从JSON文件加载股票数据"""
33        with open(file_path, 'r', encoding='utf-8') as f:
34            self.stock_data = json.load(f)
35 
36    def _assess_roe(self, roe: float) -> str:
37        """根据ROE评估盈利能力"""
38        if roe > 20:
39            return "优秀 - ROE超过20%，盈利能力很强"
40        elif roe > 15:
41            return "良好 - ROE在15-20%之间，盈利能力较强"
42        elif roe > 10:
43            return "一般 - ROE在10-15%之间，盈利能力中等"
44        return "较弱 - ROE低于10%，盈利能力需要改善"
45 
46    def analyze_profitability(self) -> Dict:
47        """盈利能力分析"""
48        result = {
49            "category": "盈利能力",
50            "metrics": {},
51            "trend": [],
52            "assessment": ""
53        }
54 
55        indicators = self.stock_data.get('financial_indicators', [])
56        if not indicators:
57            return result
58 
59        metrics = {}
60        for indicator in indicators[:8]:
61            period = indicator.get('日期', '')
62            if period:
63                metrics[period] = {
64                    "ROE": self._safe_float(indicator.get('净资产收益率', indicator.get('加权净资产收益率'))),
65                    "ROA": self._safe_float(indicator.get('总资产报酬率')),
66                    "毛利率": self._safe_float(indicator.get('销售毛利率')),
67                    "净利率": self._safe_float(indicator.get('销售净利率'))
68                }
69 
70        if not metrics:
71            return result
72 
73        latest = list(metrics.values())[0]
74        result["metrics"] = {
75            "当前ROE": latest.get("ROE"),
76            "当前ROA": latest.get("ROA"),
77            "当前毛利率": latest.get("毛利率"),
78            "当前净利率": latest.get("净利率")
79        }
80 
81        roe_values = [v.get("ROE") for v in metrics.values() if v.get("ROE") is not None]
82        if len(roe_values) >= 2:
83            trend = "上升" if roe_values[0] > roe_values[-1] else "下降"
84            result["trend"].append(f"ROE呈{trend}趋势")
85 
86        roe = latest.get("ROE")
87        if roe:
88            result["assessment"] = self._assess_roe(roe)
89 
90        return result
91 
92    def analyze_solvency(self) -> Dict:
93        """偿债能力分析"""
94        result = {
95            "category": "偿债能力",
96            "metrics": {},
97            "risks": [],
98            "assessment": ""
99        }
100 
101        indicators = self.stock_data.get('financial_indicators', [])
102        if not indicators:
103            return result
104 
105        latest = indicators[0]
106 
107        debt_ratio = self._safe_float(latest.get('资产负债率'))
108        current_ratio = self._safe_float(latest.get('流动比率'))
109        quick_ratio = self._safe_float(latest.get('速动比率'))
110 
111        result["metrics"] = {
112            "资产负债率": debt_ratio,
113            "流动比率": current_ratio,
114            "速动比率": quick_ratio
115        }
116 
117        # 风险评估
118        risk_checks = [
119            (debt_ratio and debt_ratio > 70, f"资产负债率偏高 ({debt_ratio:.1f}%)，需关注偿债压力"),
120            (current_ratio and current_ratio < 1, f"流动比率偏低 ({current_ratio:.2f})，短期偿债能力较弱"),
121            (quick_ratio and quick_ratio < 0.8, f"速动比率偏低 ({quick_ratio:.2f})，短期流动性风险"),
122        ]
123        result["risks"] = [msg for condition, msg in risk_checks if condition]
124 
125        # 综合评估
126        risk_count = len(result["risks"])
127        if risk_count == 0:
128            result["assessment"] = "良好 - 偿债能力指标正常，财务结构稳健"
129        elif risk_count == 1:
130            result["assessment"] = "一般 - 存在一项风险指标，需持续关注"
131        else:
132            result["assessment"] = "较弱 - 存在多项风险指标，偿债压力较大"
133 
134        return result
135 
136    def analyze_operation(self) -> Dict:
137        """运营效率分析"""
138        result = {
139            "category": "运营效率",
140            "metrics": {},
141            "observations": [],
142            "assessment": ""
143        }
144 
145        indicators = self.stock_data.get('financial_indicators', [])
146        if not indicators:
147            return result
148 
149        latest = indicators[0]
150 
151        ar_days = self._safe_float(latest.get('应收账款周转天数'))
152        inventory_days = self._safe_float(latest.get('存货周转天数'))
153        asset_turnover = self._safe_float(latest.get('总资产周转率'))
154 
155        result["metrics"] = {
156            "应收账款周转率": self._safe_float(latest.get('应收账款周转率')),
157            "应收账款周转天数": ar_days,
158            "存货周转率": self._safe_float(latest.get('存货周转率')),
159            "存货周转天数": inventory_days,
160            "总资产周转率": asset_turnover
161        }
162 
163        # 观察分析
164        observation_checks = [
165            (ar_days and ar_days > 90, f"应收账款周转天数较长 ({ar_days:.0f}天)，回款较慢"),
166            (inventory_days and inventory_days > 180, f"存货周转天数较长 ({inventory_days:.0f}天)，库存管理需关注"),
167            (asset_turnover and asset_turnover < 0.5, f"总资产周转率较低 ({asset_turnover:.2f})，资产利用效率有待提高"),
168        ]
169        result["observations"] = [msg for condition, msg in observation_checks if condition]
170 
171        if not result["observations"]:
172            result["assessment"] = "良好 - 运营效率指标正常"
173        else:
174            result["assessment"] = "需关注 - " + "；".join(result["observations"])
175 
176        return result
177 
178    def _assess_growth_rate(self, avg_growth: float) -> str:
179        """评估增长率"""
180        if avg_growth > 20:
181            return "高成长 - 平均增长率超过20%"
182        elif avg_growth > 10:
183            return "稳定成长 - 平均增长率10-20%"
184        elif avg_growth > 0:
185            return "低速成长 - 平均增长率0-10%"
186        return "负增长 - 需要深入分析原因"
187 
188    def _analyze_growth_trend(self, values: List[float], name: str) -> Optional[str]:
189        """分析增长趋势"""
190        if not values:
191            return None
192        if all(g > 0 for g in values[:4]):
193            return f"{name}持续正增长"
194        elif values[0] < 0:
195            return f"{name}负增长，需关注"
196        return None
197 
198    def analyze_growth(self) -> Dict:
199        """成长性分析"""
200        result = {
201            "category": "成长性",
202            "metrics": {},
203            "trend": [],
204            "assessment": ""
205        }
206 
207        indicators = self.stock_data.get('financial_indicators', [])
208        if not indicators:
209            return result
210 
211        revenue_growth = []
212        profit_growth = []
213 
214        for indicator in indicators[:8]:
215            rev = self._safe_float(indicator.get('主营业务收入增长率', indicator.get('营业收入增长率')))
216            net = self._safe_float(indicator.get('净利润增长率'))
217            if rev is not None:
218                revenue_growth.append(rev)
219            if net is not None:
220                profit_growth.append(net)
221 
222        if revenue_growth:
223            result["metrics"]["最近营收增长率"] = revenue_growth[0]
224            result["metrics"]["平均营收增长率"] = sum(revenue_growth) / len(revenue_growth)
225 
226        if profit_growth:
227            result["metrics"]["最近净利润增长率"] = profit_growth[0]
228            result["metrics"]["平均净利润增长率"] = sum(profit_growth) / len(profit_growth)
229 
230        # 趋势判断
231        for values, name in [(revenue_growth, "营收"), (profit_growth, "净利润")]:
232            trend = self._analyze_growth_trend(values, name)
233            if trend:
234                result["trend"].append(trend)
235 
236        avg_growth = result["metrics"].get("平均净利润增长率", 0) or 0
237        result["assessment"] = self._assess_growth_rate(avg_growth)
238 
239        return result
240 
241    def analyze_dupont(self) -> Dict:
242        """杜邦分析"""
243        result = {
244            "category": "杜邦分析",
245            "decomposition": {},
246            "driver": "",
247            "assessment": ""
248        }
249 
250        indicators = self.stock_data.get('financial_indicators', [])
251        if not indicators:
252            return result
253 
254        latest = indicators[0] if indicators else {}
255 
256        # 杜邦分解: ROE = 净利率 × 资产周转率 × 权益乘数
257        net_margin = self._safe_float(latest.get('销售净利率'))
258        asset_turnover = self._safe_float(latest.get('总资产周转率'))
259        equity_multiplier = self._safe_float(latest.get('权益乘数'))
260        roe = self._safe_float(latest.get('净资产收益率', latest.get('加权净资产收益率')))
261 
262        result["decomposition"] = {
263            "ROE": roe,
264            "净利率": net_margin,
265            "资产周转率": asset_turnover,
266            "权益乘数": equity_multiplier
267        }
268 
269        # 判断ROE驱动因素
270        if net_margin and asset_turnover and equity_multiplier:
271            drivers = []
272            if net_margin > 15:
273                drivers.append("高净利率")
274            if asset_turnover > 1:
275                drivers.append("高周转")
276            if equity_multiplier > 2.5:
277                drivers.append("高杠杆")
278 
279            if drivers:
280                result["driver"] = "ROE主要由" + "、".join(drivers) + "驱动"
281            else:
282                result["driver"] = "ROE驱动因素较为均衡"
283 
284        return result
285 
286    def detect_anomalies(self) -> Dict:
287        """财务异常检测"""
288        result = {
289            "category": "财务异常检测",
290            "signals": [],
291            "risk_level": "低",
292            "details": []
293        }
294 
295        indicators = self.stock_data.get('financial_indicators', [])
296        if len(indicators) < 2:
297            return result
298 
299        current = indicators[0]
300        previous = indicators[1] if len(indicators) > 1 else {}
301 
302        # 1. 应收账款异常
303        ar_growth = self._safe_float(current.get('应收账款增长率'))
304        revenue_growth = self._safe_float(current.get('营业收入增长率', current.get('主营业务收入增长率')))
305        if ar_growth and revenue_growth and ar_growth > revenue_growth * 1.5:
306            result["signals"].append({
307                "type": "应收账款增速异常",
308                "description": f"应收账款增速({ar_growth:.1f}%)显著高于营收增速({revenue_growth:.1f}%)",
309                "severity": "中"
310            })
311 
312        # 2. 存货异常
313        inventory_growth = self._safe_float(current.get('存货增长率'))
314        if inventory_growth and revenue_growth and inventory_growth > revenue_growth * 2:
315            result["signals"].append({
316                "type": "存货增速异常",
317                "description": f"存货增速({inventory_growth:.1f}%)远高于营收增速({revenue_growth:.1f}%)",
318                "severity": "中"
319            })
320 
321        # 3. 毛利率异常波动
322        current_gm = self._safe_float(current.get('销售毛利率'))
323        previous_gm = self._safe_float(previous.get('销售毛利率'))
324        if current_gm and previous_gm:
325            gm_change = abs(current_gm - previous_gm)
326            if gm_change > 10:
327                result["signals"].append({
328                    "type": "毛利率大幅波动",
329                    "description": f"毛利率变动{gm_change:.1f}个百分点，需关注原因",
330                    "severity": "中"
331                })
332 
333        # 4. 经营现金流与净利润背离 (需要现金流数据)
334        cash_flow = self.stock_data.get('financial_data', {}).get('cash_flow', [])
335        income = self.stock_data.get('financial_data', {}).get('income_statement', [])
336        if cash_flow and income:
337            try:
338                ocf = self._safe_float(cash_flow[0].get('经营活动产生的现金流量净额'))
339                net_profit = self._safe_float(income[0].get('净利润'))
340                if ocf and net_profit and net_profit > 0:
341                    ocf_ratio = ocf / net_profit
342                    if ocf_ratio < 0.5:
343                        result["signals"].append({
344                            "type": "现金流与利润背离",
345                            "description": f"经营现金流/净利润 = {ocf_ratio:.1%}，盈利质量存疑",
346                            "severity": "高"
347                        })
348            except (IndexError, KeyError):
349                pass
350 
351        # 确定风险等级
352        high_severity = sum(1 for s in result["signals"] if s.get("severity") == "高")
353        medium_severity = sum(1 for s in result["signals"] if s.get("severity") == "中")
354 
355        if high_severity > 0:
356            result["risk_level"] = "高"
357        elif medium_severity >= 2:
358            result["risk_level"] = "中"
359        else:
360            result["risk_level"] = "低"
361 
362        return result
363 
364    def generate_summary(self, level: str = "standard") -> Dict:
365        """生成分析摘要"""
366        summary = {
367            "code": self.stock_data.get('code', ''),
368            "name": self.stock_data.get('basic_info', {}).get('name', ''),
369            "analysis_date": datetime.now().isoformat(),
370            "level": level
371        }
372 
373        profitability = self.analyze_profitability()
374        solvency = self.analyze_solvency()
375        operation = self.analyze_operation()
376        growth = self.analyze_growth()
377        anomalies = self.detect_anomalies()
378 
379        if level == "summary":
380            summary["profitability"] = profitability["assessment"]
381            summary["solvency"] = solvency["assessment"]
382            summary["growth"] = growth["assessment"]
383            summary["risk_level"] = anomalies["risk_level"]
384        else:
385            # standard 和 deep 级别共享基础分析结果
386            summary["profitability"] = profitability
387            summary["solvency"] = solvency
388            summary["operation"] = operation
389            summary["growth"] = growth
390            summary["dupont"] = self.analyze_dupont()
391            summary["anomalies"] = anomalies
392 
393            if level == "deep":
394                summary["historical_indicators"] = self.stock_data.get('financial_indicators', [])
395 
396        summary["score"] = self._calculate_score(profitability, solvency, growth, anomalies)
397 
398        return summary
399 
400    def _calculate_score(self, profitability, solvency, growth, anomalies) -> int:
401        """计算综合评分 (0-100)"""
402        score = 50
403 
404        # 盈利能力评分
405        roe = profitability.get("metrics", {}).get("当前ROE")
406        if roe:
407            if roe > 20:
408                score += 15
409            elif roe > 15:
410                score += 10
411            elif roe > 10:
412                score += 5
413            elif roe < 5:
414                score -= 5
415 
416        # 偿债能力评分
417        risks = solvency.get("risks", [])
418        score += 10 if not risks else -len(risks) * 3
419 
420        # 成长性评分
421        avg_growth = growth.get("metrics", {}).get("平均净利润增长率", 0) or 0
422        if avg_growth > 20:
423            score += 15
424        elif avg_growth > 10:
425            score += 10
426        elif avg_growth > 0:
427            score += 5
428        else:
429            score -= 5
430 
431        # 风险扣分
432        risk_penalties = {"高": 15, "中": 8}
433        score -= risk_penalties.get(anomalies.get("risk_level", "低"), 0)
434 
435        return max(0, min(100, score))
436 
437    def compare_stocks(self, stocks_data: List[Dict]) -> Dict:
438        """对比多只股票"""
439        comparison = {
440            "comparison_date": datetime.now().isoformat(),
441            "stocks": [],
442            "ranking": {}
443        }
444 
445        stock_scores = []
446        for stock in stocks_data:
447            self.stock_data = stock
448            summary = self.generate_summary(level="summary")
449            comparison["stocks"].append({
450                "code": stock.get('code', ''),
451                "name": stock.get('basic_info', {}).get('name', ''),
452                "score": summary.get("score", 50),
453                "profitability": summary.get("profitability"),
454                "solvency": summary.get("solvency"),
455                "growth": summary.get("growth"),
456                "risk_level": summary.get("risk_level")
457            })
458            stock_scores.append((stock.get('code', ''), summary.get("score", 50)))
459 
460        # 排名
461        stock_scores.sort(key=lambda x: x[1], reverse=True)
462        comparison["ranking"] = {code: rank + 1 for rank, (code, _) in enumerate(stock_scores)}
463 
464        return comparison
465 
466    @staticmethod
467    def _safe_float(value) -> Optional[float]:
468        """安全转换为浮点数"""
469        if value is None or value == '' or value == '--':
470            return None
471        try:
472            if isinstance(value, str):
473                value = value.replace('%', '').replace(',', '')
474            return float(value)
475        except (ValueError, TypeError):
476            return None
477 
478 
479def main():
480    parser = argparse.ArgumentParser(description="A股财务分析器")
481    parser.add_argument("--input", type=str, required=True, help="输入数据文件 (JSON)")
482    parser.add_argument("--level", type=str, default="standard",
483                       choices=["summary", "standard", "deep"],
484                       help="分析深度级别")
485    parser.add_argument("--mode", type=str, default="single",
486                       choices=["single", "comparison"],
487                       help="分析模式: single(单只)/comparison(对比)")
488    parser.add_argument("--output", type=str, help="输出文件路径 (JSON)")
489 
490    args = parser.parse_args()
491 
492    # 加载数据
493    with open(args.input, 'r', encoding='utf-8') as f:
494        data = json.load(f)
495 
496    analyzer = FinancialAnalyzer()
497 
498    if args.mode == "single":
499        analyzer.stock_data = data
500        result = analyzer.generate_summary(level=args.level)
501    else:
502        # 对比模式
503        stocks = data.get('stocks', [data])
504        result = analyzer.compare_stocks(stocks)
505 
506    # 输出
507    output_json = json.dumps(result, ensure_ascii=False, indent=2, default=str)
508 
509    if args.output:
510        with open(args.output, 'w', encoding='utf-8') as f:
511            f.write(output_json)
512        print(f"分析结果已保存到: {args.output}")
513    else:
514        print(output_json)
515 
516 
517if __name__ == "__main__":
518    main()
519