Python + 同花顺问财：实战——用排雷工具箱批量扫描全A排雷

前面把二级系列 12-16 打包成了【二级市场进阶排雷工具箱（二）】，五个 Python 评分卡函数一次备齐。但工具箱的价值不在"看一眼"，在"批量跑"。一只只人工翻财报、查北向、盯减持，全A五千多只根本看不过来。这篇直接上实战：把五个函数接上批量扫描，跑出一份"该重点查"的排雷清单。

为什么必须批量扫（注师视角）

完整 scanner，零依赖（仅 csv + 标准库），保存为 scan_toolbox2.py：

# -*- coding: utf-8 -*-
# Full-A scanner for 二级市场进阶排雷工具箱（二）
# Zero external dependency: only csv + stdlib.
# Usage:
#   python scan_toolbox2.py            # run with built-in demo universe
#   python scan_toolbox2.py data.csv   # scan your own exported universe
# CSV columns expected:
#   code,name,gross_margin,gm_change,industry_avg,
#   north_holding,north_change_1m,roe,
#   reduce_ratio,net_growth,reduce_amount,
#   convert_premium,purebond_premium,time_to_put,
#   dividend_yield,payout_3y,buyback,fcf

import csv
import random
import sys

def moat_signal(gross_margin, gm_change, industry_avg):
    s = 0
    if gm_change < -0.05:
        s += 1
    if gross_margin > industry_avg:
        s += 1
    return s

def north_bound_signal(holding_ratio, change_1m):
    s = 0
    if holding_ratio > 0.03:
        s += 1
    if change_1m > 0.20:
        s += 1
    return s

def insider_signal(reduce_ratio, net_growth, reduce_amount):
    s = 0
    if reduce_ratio > 0.01:
        s += 1
    if net_growth > 0:
        s += 1
    if reduce_amount > 0:
        s += 1
    return s

def cb_signal(convert_premium, purebond_premium, time_to_put):
    s = 0
    if convert_premium > 0.30:
        s += 1
    if purebond_premium < 0.10:
        s += 1
    if time_to_put < 1:
        s += 1
    return s

def payout_signal(dividend_yield, payout_ratio_3y, buyback, fcf):
    s = 0
    if dividend_yield > 0.02:
        s += 1
    if payout_ratio_3y > 0.30:
        s += 1
    if buyback > 0 and fcf > 0:
        s += 1
    return s

SIGNAL_NAMES = ["moat", "north", "insider", "cb", "payout"]

def score_row(r):
    s1 = moat_signal(r["gross_margin"], r["gm_change"], r["industry_avg"])
    s2 = north_bound_signal(r["north_holding"], r["north_change_1m"])
    s3 = insider_signal(r["reduce_ratio"], r["net_growth"], r["reduce_amount"])
    s4 = cb_signal(r["convert_premium"], r["purebond_premium"], r["time_to_put"])
    s5 = payout_signal(r["dividend_yield"], r["payout_3y"], r["buyback"], r["fcf"])
    return [s1, s2, s3, s4, s5]

def load_csv(path):
    rows = []
    with open(path, newline="", encoding="utf-8-sig") as f:
        for d in csv.DictReader(f):
            rows.append({k: _to_float(v) for k, v in d.items()})
    return rows

def _to_float(v):
    try:
        return float(v)
    except (TypeError, ValueError):
        return 0.0

def demo_universe(n=60, seed=20260822):
    random.seed(seed)
    known = [
        ("600519", "贵州茅台"), ("000858", "五粮液"), ("300750", "宁德时代"),
        ("601318", "中国平安"), ("000001", "平安银行"), ("600036", "招商银行"),
        ("000333", "美的集团"), ("000651", "格力电器"), ("600276", "恒瑞医药"),
        ("002594", "比亚迪"), ("600900", "长江电力"), ("601012", "隆基绿能"),
        ("000725", "京东方A"), ("600030", "中信证券"), ("601888", "中国中免"),
    ]
    rows = []
    for i in range(n):
        if i < len(known):
            code, name = known[i]
        else:
            code = "%06d" % (600000 + i)
            name = "stock%d" % i
        rows.append({
            "code": code, "name": name,
            "gross_margin": round(random.uniform(0.10, 0.70), 3),
            "gm_change": round(random.uniform(-0.15, 0.15), 3),
            "industry_avg": round(random.uniform(0.20, 0.45), 3),
            "north_holding": round(random.uniform(0.0, 0.08), 3),
            "north_change_1m": round(random.uniform(-0.1, 0.5), 3),
            "roe": round(random.uniform(0.02, 0.25), 3),
            "reduce_ratio": round(random.uniform(0.0, 0.04), 3),
            "net_growth": round(random.uniform(-0.2, 0.3), 3),
            "reduce_amount": round(random.uniform(0, 8000), 1),
            "convert_premium": round(random.uniform(0.0, 0.6), 3),
            "purebond_premium": round(random.uniform(0.0, 0.2), 3),
            "time_to_put": round(random.uniform(0.2, 3.0), 2),
            "dividend_yield": round(random.uniform(0.0, 0.06), 3),
            "payout_3y": round(random.uniform(0.1, 0.9), 3),
            "buyback": round(random.uniform(0, 3000), 1),
            "fcf": round(random.uniform(-2000, 8000), 1),
        })
    return rows

def scan(rows):
    out = []
    for r in rows:
        scores = score_row(r)
        total = sum(scores)
        rec = {"code": r["code"], "name": r["name"]}
        for nm, sc in zip(SIGNAL_NAMES, scores):
            rec[nm] = sc
        rec["total"] = total
        out.append(rec)
    out.sort(key=lambda x: x["total"], reverse=True)
    return out

def report(results, top=15):
    print("=== Top %d by total risk score (0-12) ===" % top)
    for rec in results[:top]:
        flag = "".join(
            [nm[0].upper() for nm, sc in zip(SIGNAL_NAMES, [rec[n] for n in SIGNAL_NAMES]) if sc > 0]
        )
        print("%s %-10s total=%d flags=%s" % (rec["code"], rec["name"], rec["total"], flag))
    with open("scan_result.csv", "w", newline="", encoding="utf-8-sig") as f:
        w = csv.writer(f)
        w.writerow(["code", "name"] + SIGNAL_NAMES + ["total"])
        for rec in results:
            w.writerow([rec["code"], rec["name"]] + [rec[n] for n in SIGNAL_NAMES] + [rec["total"]])
    print("\nFull result saved to scan_result.csv (%d stocks)" % len(results))

def main():
    if len(sys.argv) > 1:
        rows = load_csv(sys.argv[1])
        print("Loaded %d stocks from %s" % (len(rows), sys.argv[1]))
    else:
        rows = demo_universe()
        print("No CSV given, running built-in demo universe (%d stocks, simulated values)" % len(rows))
    results = scan(rows)
    report(results)

if __name__ == "__main__":
    main()


CSV 列名（顺序不限，脚本按列名取）：

示例输出（模拟数据，仅演示）：

=== Top 15 by total risk score (0-12) ===
600900 长江电力   total=11 flags=MNICP
300750 宁德时代   total=10 flags=MNICP
601318 中国平安   total=10 flags=MNICP


注意：示例是随机模拟值，只为展示流程；接真实数据后分数才有区分度。

工具箱（二）五个函数 + 批量 scanner = 一键把全A缩成"待查清单"。问财管筛池，Python 管打分，注师管定性收口。量化是放大镜，不是法官。

一个信号亮灯，可能只是噪声——北向一次调仓、毛利率一季波动，单看都不足以定罪（先验很低）。但当毛利率、北向、减持、转债、回报五个相互独立的角度同时亮灯，按贝叶斯公式，目标公司真有问题的后验概率会陡然上升。批量扫描的本质，就是低成本的"证据累加器"：它不替你下结论，而是把职业怀疑精准地投到后验概率最高的那一批上。把筛选交给机器，把判断留给人——这才是注师与量化该有的分工。

觉得这套扫描实战能帮你把全A缩成待查清单，点赞收藏 + 关注。下期把二级系列 1-11 的问财公式和 Python 评分卡打包成【二级市场进阶排雷工具箱（一）】放出，从 DCF 到回购分红全谱系排雷一次备齐。会计师事务所的一名注册会计师

编程语言

人工智能

Python+同花顺问财实战：用排雷工具箱批量扫描全A股，五维评分卡（毛利率、北向、减持、转债、分红）自动打分，零依赖代码一键生成“重点核查清单”，将5000只股票精准缩至50只待查池，提升审计效率。

Python

2026上云采购 | AI焕新·智启新局

cloudaudit

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

Python + 同花顺问财：实战——用排雷工具箱批量扫描全A排雷

Python + 同花顺问财：实战——用排雷工具箱批量扫描全A排雷

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐