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 解决方案（用户接受）

两步法：

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

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

关键代码骨架

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 算法（已废，但记录"为什么不自己算"）

7.9 KiB Raw Blame History Unescape Escape