如何像分析师一样利用 AI 深度解读任何公司

1#!/usr/bin/env python3
2"""
3forensic_screener.py  - 像分析师一样阅读任何公司，一条命令搞定。
4
5setup:
6    pip install edgartools anthropic
7    export SEC_IDENTITY="Your Name [email protected]"   # SEC 要求此标头
8    export ANTHROPIC_API_KEY="sk-..."                # 可选，仅用于 diff
9
10run:
11    python forensic_screener.py AAPL
12    python forensic_screener.py TSLA NVDA SMCI       # 同时筛选多个
13    python forensic_screener.py SMCI --diff          # 添加风险因素 diff
14"""
15
16import os, sys, argparse
17from dataclasses import dataclass
18
19# 阈值 - 改变你立场的标杆。可根据喜好调整。
20M_FLAG      = -1.78    # Beneish 指标高于此值 -> 操纵风险（经典临界值为 -2.22）
21Z_DISTRESS  =  1.81    # Altman 指标低于此值 -> 困境区域
22Z_SAFE      =  2.99    # Altman 指标高于此值 -> 安全区域
23ACCRUAL_BAD =  0.25    # |应计/资产| 高于此值 -> 盈利质量红旗
24F_STRONG    =  6       # Piotroski 指标达到或高于此值 -> 正在加强
25
26@dataclass
27class YearData:
28    sales: float; cogs: float; sga: float; net_income: float; cfo: float
29    receivables: float; current_assets: float; current_liabilities: float
30    ppe_net: float; depreciation: float; total_assets: float
31    total_liabilities: float; long_term_debt: float; retained_earnings: float
32    ebit: float; market_cap: float; shares: float
33
34def load_two_years(ticker: str):
35    """returns (this_year, last_year). 通过网络调用 SEC EDGAR。"""
36    from edgar import Company, set_identity
37    identity = os.environ.get("SEC_IDENTITY")
38    if not identity:
39        sys.exit("设置 SEC_IDENTITY='Your Name [email protected]' - SEC 要求此项。")
40    set_identity(identity)
41
42    company = Company(ticker)
43    fin = company.get_financials()
44    inc, bal, cfs = fin.income_statement(periods=2), fin.balance_sheet(periods=2), fin.cash_flow(periods=2)
45
46    def g(stmt, col, *aliases):
47        # 尽力而为的行查找；公司对同一概念的标记方式不同
48        for a in aliases:
49            try:
50                row = stmt.loc[stmt.index.str.contains(a, case=False, na=False)]
51                if not row.empty:
52                    return float(row.iloc[0, col])
53            except Exception:
54                continue
55        return 0.0
56
57    def build(col):
58        return YearData(
59            sales=g(inc, col, "RevenueFromContractWithCustomerExcludingAssessedTax", "Revenues", "SalesRevenueNet"),
60            cogs=g(inc, col, "CostOfGoodsAndServicesSold", "CostOfRevenue"),
61            sga=g(inc, col, "SellingGeneralAndAdministrativeExpense"),
62            net_income=g(inc, col, "NetIncomeLoss"),
63            cfo=g(cfs, col, "NetCashProvidedByUsedInOperatingActivities"),
64            receivables=g(bal, col, "AccountsReceivableNetCurrent", "ReceivablesNetCurrent"),
65            current_assets=g(bal, col, "AssetsCurrent"),
66            current_liabilities=g(bal, col, "LiabilitiesCurrent"),
67            ppe_net=g(bal, col, "PropertyPlantAndEquipmentNet"),
68            depreciation=g(cfs, col, "DepreciationDepletionAndAmortization", "DepreciationAmortizationAndAccretionNet"),
69            total_assets=g(bal, col, "Assets"),
70            total_liabilities=g(bal, col, "Liabilities"),
71            long_term_debt=g(bal, col, "LongTermDebtNoncurrent", "LongTermDebt"),
72            retained_earnings=g(bal, col, "RetainedEarningsAccumulatedDeficit"),
73            ebit=g(inc, col, "OperatingIncomeLoss"),
74            market_cap=getattr(company, "market_cap", 0.0) or 0.0,
75            shares=g(bal, col, "CommonStockSharesOutstanding"),
76        )
77
78    # col 0 = 最新年份，col 1 = 前一年（edgartools 将最新的排在前面）
79    return build(0), build(1)
80
81def d(a, b): return a / b if b else 0.0   # 安全除法
82
83def beneish_m_score(t, p):
84    DSRI = d(d(t.receivables, t.sales), d(p.receivables, p.sales))
85    GMI  = d((p.sales - p.cogs)/p.sales if p.sales else 0, (t.sales - t.cogs)/t.sales if t.sales else 0)
86    AQI  = d(1 - d(t.current_assets + t.ppe_net, t.total_assets), 1 - d(p.current_assets + p.ppe_net, p.total_assets))
87    SGI  = d(t.sales, p.sales)
88    DEPI = d(d(p.depreciation, p.depreciation + p.ppe_net), d(t.depreciation, t.depreciation + t.ppe_net))
89    SGAI = d(d(t.sga, t.sales), d(p.sga, p.sales))
90    TATA = d(t.net_income - t.cfo, t.total_assets)
91    LVGI = d(d(t.total_liabilities, t.total_assets), d(p.total_liabilities, p.total_assets))
92    return (-4.84 + 0.92*DSRI + 0.528*GMI + 0.404*AQI + 0.892*SGI
93            + 0.115*DEPI - 0.172*SGAI + 4.679*TATA - 0.327*LVGI)
94
95def altman_z_score(t):
96    wc = t.current_assets - t.current_liabilities
97    return (1.2*d(wc, t.total_assets) + 1.4*d(t.retained_earnings, t.total_assets)
98            + 3.3*d(t.ebit, t.total_assets) + 0.6*d(t.market_cap, t.total_liabilities)
99            + 1.0*d(t.sales, t.total_assets))
100
101def piotroski_f_score(t, p):
102    s = 0
103    s += t.net_income > 0
104    s += t.cfo > 0
105    s += d(t.net_income, t.total_assets) > d(p.net_income, p.total_assets)
106    s += t.cfo > t.net_income                                  # 现金超越应计
107    s += t.long_term_debt < p.long_term_debt
108    s += d(t.current_assets, t.current_liabilities) > d(p.current_assets, p.current_liabilities)
109    s += t.shares <= p.shares                                  # 无稀释
110    s += d(t.sales - t.cogs, t.sales) > d(p.sales - p.cogs, p.sales)
111    s += d(t.sales, t.total_assets) > d(p.sales, p.total_assets)
112    return int(s)
113
114def sloan_accruals(t): return d(t.net_income - t.cfo, t.total_assets)
115
116def risk_factor_diff(ticker):
117    """对比今年与去年的风险因素。整个分析栈中最佳单一操作。"""
118    key = os.environ.get("ANTHROPIC_API_KEY")
119    if not key:
120        return "(已跳过 - 设置 ANTHROPIC_API_KEY 以启用 diff)"
121    from edgar import Company
122    import anthropic
123    f = Company(ticker).get_filings(form="10-K").latest(2)
124    this_rf = getattr(f[0].obj(), "risk_factors", str(f[0]))
125    last_rf = getattr(f[1].obj(), "risk_factors", str(f[1]))
126    client = anthropic.Anthropic(api_key=key)
127    msg = client.messages.create(
128        model="claude-opus-4-8", max_tokens=1024,
129        messages=[{"role": "user", "content": (
130            "比较来自连续年度申报文件的两个风险因素章节。"
131            "仅报告今年 NEW 或今年 REMOVED 的内容。引用新语言。"
132            "忽略两者都有的模板化内容。用一句话结束：这里的任何内容是否会改变风险？\n\n"
133            f"去年：\n{last_rf[:40000]}\n\n今年：\n{this_rf[:40000]}")}],
134    )
135    return msg.content[0].text
136
137def screen(ticker, do_diff=False):
138    print(f"\n{'='*60}\n  {ticker.upper()}\n{'='*60}")
139    try:
140        t, p = load_two_years(ticker)
141    except Exception as e:
142        print(f"  无法加载申报文件：{e}"); return
143
144    m, z, f, a = beneish_m_score(t, p), altman_z_score(t), piotroski_f_score(t, p), sloan_accruals(t)
145    flags = []
146    if m > M_FLAG:           flags.append(f"M-Score {m:+.2f} - 盈利操纵风险")
147    if z < Z_DISTRESS:       flags.append(f"Z-Score {z:.2f} - 财务困境区域")
148    if abs(a) > ACCRUAL_BAD: flags.append(f"应计项目 {a:+.1%} - 盈利质量红旗")
149    if f < F_STRONG:         flags.append(f"F-Score {f}/9 - 未在加强")
150
151    print(f"  Beneish M  : {m:+.2f}   (> {M_FLAG} = 需要调查)")
152    print(f"  Altman Z   : {z:.2f}    (< {Z_DISTRESS} 困境, > {Z_SAFE} 安全)")
153    print(f"  Piotroski F: {f}/9      (>= {F_STRONG} 强)")
154    print(f"  Sloan Accr : {a:+.1%}   (|x| > {ACCRUAL_BAD:.0%} 红旗)")
155    print(f"\n  结论：{'需要调查' if flags else '状况良好'}")
156    for fl in flags: print(f"    - {fl}")
157    if do_diff:
158        print("\n  风险因素对比 (年度间):")
159        print("  " + risk_factor_diff(ticker).replace("\n", "\n  "))
160
161def main():
162    ap = argparse.ArgumentParser(description="针对任何美国上市公司的法务筛选器")
163    ap.add_argument("tickers", nargs="+")
164    ap.add_argument("--diff", action="store_true", help="同时运行风险因素对比")
165    args = ap.parse_args()
166    for tk in args.tickers:
167        screen(tk, do_diff=args.diff)
168    print("\n提醒：这是概率信号，不是确凿证据。Beneish M-Score 基于去年数据运行，因此"
169          "操纵可能已经开始反转。一个糟糕的分数意味着需要打开申报文件，切勿仅根据这个数字就做空。\n")
170
171if __name__ == "__main__":
172    main()