# A 股 Top N 高息扫描 - v5 周批量化版（2026-09-19 实装）

## 背景

之前的 A 股高息扫描全靠 `cn_dividend_buy_timing.py` 硬编码 8 只候选池（601288/601398/601088/600023/600585/601006/601939/601658），用户要求改为自动从全市场筛 Top N。

v1-v4 失败链路：
- v1: 用 akshare `stock_history_dividend_detail` 自己算 TTM → 单位错了（/10 vs 每股）
- v3: 用东方财富 `现金分红-股息率`（小数）→ 字段单位忘了 `* 100`，差 27 倍
- v4: 改用财年滚动算法 + 含预案 → 算到 5.44%，同花顺显示 7.36%，**还是不准**

**用户原话**: "你一个AI写算法，还能比得过人家真正的专家吗？... 我也不专业，算出来我也不知道对不对，我也是去看同花顺的"

**核心教训**: **agent 算法永远追不上同花顺/东方财富专业团队** → **直接用同花顺的接口拿数据**，不自己算。

## v5 解决方案（用户接受）

**两步法**：
1. **东方财富 `stock_fhps_em(date)` 批量粗筛**（3 秒/季，1 拉 3600+ 只，1 次拿全部候选）
2. **同花顺 `stock_fhps_detail_ths(symbol)` 并行校验**（0.5s/只，8 线程并行，100 只 7.7s）

**总耗时**：~25 秒（东方财富 15s + 同花顺 8s + 长桥价 2s）

## 关键代码骨架

```python
import akshare as ak
from concurrent.futures import ThreadPoolExecutor
import os

# 0. 关代理（akshare 国内接口被全局 proxy 干扰）
for k in ['http_proxy', 'https_proxy', 'HTTP_PROXY', 'HTTPS_PROXY']:
    os.environ.pop(k, None)

REPORT_DATES = ['20231231', '20240630', '20241231', '20250630', '20251231', '20260630']

# 1. 东方财富批量粗筛（先 TTM 365 天滚动）
def fetch_eastmoney_batch():
    all_dfs = []
    for date in REPORT_DATES:
        df = ak.stock_fhps_em(date=date)  # 1 次 3600+ 只
        df['_date'] = date
        all_dfs.append(df)
    return pd.concat(all_dfs, ignore_index=True)

def calc_ttm_yield(df_all):
    df_all['code'] = df_all['代码'].astype(str).str.zfill(6)
    df_all['yld_num'] = pd.to_numeric(df_all['现金分红-股息率'], errors='coerce') * 100  # 字段是小数!
    df_done = df_all[df_all['方案进度'] == '实施分配']
    df_done['reg_date'] = pd.to_datetime(df_done['股权登记日'], errors='coerce')
    now = datetime.now()
    ttm_start = now - timedelta(days=365)
    df_ttm = df_done[df_done['reg_date'] >= ttm_start]
    return df_ttm.groupby('code')['yld_num'].sum().to_dict(), \
           df_done.groupby('code').size().to_dict()

# 2. 同花顺并行校验（取粗筛 top 100）
def fetch_ths_one(code):
    df = ak.stock_fhps_detail_ths(symbol=code)
    # 算连续派息年数（按实施公告日）
    if '实施公告日' in df.columns:
        df['year'] = pd.to_datetime(df['实施公告日'], errors='coerce').dt.year
        years = sorted(df['year'].dropna().unique(), reverse=True)
        cont = 1
        for i in range(1, len(years)):
            if years[i-1] - years[i] == 1: cont += 1
            else: break
    # TTM = 实施公告日 365 天内税前分红率 sum
    if '税前分红率' in df.columns:
        now = datetime.now()
        ttm = now - timedelta(days=365)
        recent = df[pd.to_datetime(df['实施公告日'], errors='coerce') >= ttm]
        yld_str = recent['税前分红率'].astype(str).str.replace('%', '', regex=False)
        ttm_yield = float(pd.to_numeric(yld_str, errors='coerce').dropna().sum())
    else:
        ttm_yield = 0
    return code, cont, ttm_yield

def ths_parallel_verify(candidates, workers=8):
    results = {}
    with ThreadPoolExecutor(max_workers=workers) as ex:
        futures = {ex.submit(fetch_ths_one, code): code for code in candidates}
        for fut in as_completed(futures):
            code, cont, yld = fut.result()
            results[code] = (cont, yld)
    return results

# 3. 过滤 + 排序
def filter_and_rank(df_all, ttm_map, cnt_map, ths_results, prices):
    name_map = df_all.groupby('代码')['名称'].first().to_dict()
    rows = []
    for code, em_yield in ttm_map.items():
        name = str(name_map.get(code, ''))
        if 'ST' in name or '*ST' in name: continue
        if cnt_map.get(code, 0) < 5: continue
        cont, ths_yield = ths_results.get(code, (0, 0))
        if cont < 5: continue
        # 同花顺优先（它才是权威），没查到用东方财富
        final_yield = ths_yield if ths_yield > 0 else em_yield
        rows.append({
            'code': code, 'name': name,
            'yield_pct': round(final_yield, 2),
            'continuous': cont,
            'div_count': cnt_map.get(code, 0),
            'price': prices.get(code, 0),
            'data_source': 'ths' if ths_yield > 0 else 'em',
        })
    rows.sort(key=lambda x: -x['yield_pct'])
    return rows[:TOP_N]
```

## 实战数据 (2026-09-19 跑通)

**Top 8**（全部用同花顺 TTM 数据源）：

| 代码 | 名称 | TTM 股息率 | 连续(年) | 数据源 |
|---|---|---|---|---|
| 603165 | 荣晟环保 | 13.45% | 10 | ths |
| 000915 | 华特达因 | 8.6% | 14 | ths |
| 603551 | 奥普科技 | 7.81% | 7 | ths |
| 300406 | 九强生物 | 7.08% | 12 | ths |
| 002867 | 周大生 | 6.96% | 10 | ths |
| 601919 | 中远海控 | 6.9% | 6 | ths |
| 600015 | 华夏银行 | 6.06% | 24 | ths |
| 600750 | 华润江中 | 5.88% | 26 | ths |

**耗时**: 25.3 秒（东方财富 14.5s + 同花顺并行 7.7s + 长桥 3s）

## 关键踩坑

| 坑 | 解决 |
|---|---|
| **东方财富 `现金分红-股息率` 是小数（0.0544）不是字符串** | `* 100` 转 % |
| **同花顺 `税前分红率` 是字符串（"5.44%"）** | `.str.replace('%', '')` + `float()` |
| **akshare 国内接口被全局 HTTP_PROXY 拦截** | 跑脚本前 `unset` 4 个 proxy env |
| **北京银行 5.44% vs 同花顺 7.36%** | **财年滚动**算法 + 含预案 = **正解 7.44%** |
| **同花顺 ths detail 没批量接口** | ThreadPoolExecutor 并行（8 线程够稳，16 触发风控） |
| **拉 5675 只价格超 10 分钟** | 粗筛 top 100 后只拉 100 只（长桥 batch 内部 15/批） |

## Cron 集成（待 user 决策）

| 项 | 现状 |
|---|---|
| **脚本路径** | `/tmp/cn_dividend_top8_v5.py`（测试版） |
| **正式版** | 待移到 `~/.hermes/scripts/cn_dividend_top8.py` |
| **cron 频率** | 每周一（用户偏好一周跑一次） |
| **缓存路径** | `~/.hermes/trading/cn_top8.json` |
| **缓存格式** | `[{"code": "601169", "name": "...", "yield_pct": 7.36, "continuous": 19, ...}]` |
| **下游消费** | `cn-dividend-buy-timing` cron 改成读 JSON，不再用 `HIGH_DIVIDEND_POOL` 硬编码 |
| **行业分散** | **不做**（akshare 行业接口代理不稳，且用户未强求） |

## 同花顺 ths 接口 vs akshare ths 接口对比

| 项 | akshare `stock_fhps_detail_ths` | 同花顺 ths 官方 API |
|---|---|---|
| 可用 | ✅ akshare 封装 | ❌ 无官方 Linux 客户端 |
| 速度 | 0.5-0.9 秒/只 | N/A |
| 字段 | 报告期 / 实施公告日 / 税前分红率 / 方案进度 | 同上 |
| 用途 | 数据验证 / 批量校验 | N/A |

**结论**：akshare ths 接口**字段口径与同花顺 APP 完全一致**——直接用这个。

## 反模式（用户骂过的）

- ❌ 自己写 TTM 算法 → "你一个AI写算法，还能比得过专家吗？"
- ❌ 反复 retry 同花顺接口 / akshare 接口 → "你是不是 API 没调通？"
- ❌ 用 `summary['年均股息'] / 当前价` 当股息率 → 错 27 倍（漏 `/ 10`）
- ❌ 用 `派息 / 当前价` 当股息率 → 错 10 倍（漏 `/ 10`）
- ❌ 5 次算法调试 vs 1 行接口验证 → 用户骂 "我不专业，算出来我也不知道对不对，我也是去看同花顺的"

## 关联 skill

- `~/.hermes/skills/prefer-data-over-computation/SKILL.md` —— 算法 vs 源数据原则
- `~/.hermes/skills/long-toolchain-pacing/references/akshare-a-share-dividend.md` —— akshare 接口细节 + 单位陷阱
- `~/.hermes/skills/trading/dividend-scanner/references/cn-top8-auto-screen.md` —— v4 算法（已废，但记录"为什么不自己算"）