A 股 AI 选股 Agent 完整工作流

结论：模型负责复核，不负责从零枚举股票

最小可用 Agent 由四个部分组成：带时间戳的数据快照、确定性候选过滤、基于证据的模型复核、样本外回测。规则先剔除无法交易、数据不足或不符合研究条件的标的，模型只阅读较小候选集。 这种分工能解释“为什么入选”，也能在模型不可用时继续运行数据和规则层。第一版只输出研究清单，不持有券商写权限。

真实项目界面展示每日研究任务

daily_stock_analysis 官方截图展示任务状态和报告入口，适合作为每日研究工作流的产品参考。本文的选股流程是独立设计，不把该截图当成本流程已运行的证明。

从全市场到模拟观察的六阶段数据流

股票池先按交易状态和数据完整性清理，再执行明确规则；模型读取候选证据卡并寻找风险；回测只使用当时可见数据；通过后进入模拟观察和人工复核。

先用代码做可解释的确定性过滤

过滤条件应来自明确研究假设，字段必须带 as_of 日期。下面示例只演示结构，不代表推荐阈值；真实阈值需要在训练窗口定义并在样本外评估。

from dataclasses import dataclass

@dataclass
class Snapshot:
    symbol: str
    as_of: str
    tradable: bool
    history_days: int
    positive_operating_cashflow: bool
    debt_ratio: float

def eligible(row: Snapshot) -> bool:
    return all([
        row.tradable,
        row.history_days >= 250,
        row.positive_operating_cashflow,
        row.debt_ratio < 0.70,
    ])

candidates = [row for row in snapshots if eligible(row)]

模型审阅证据卡并主动寻找反证

每个候选只给模型结构化指标、公告摘要、来源定位和缺失项。要求模型输出支持证据、反方证据、未知项和引用 ID，而不是单一分数。

import json, os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["KKAIAPI_KEY"],
    base_url="https://api.kkaiapi.com/v1",
)
reply = client.chat.completions.create(
    model=os.environ["KKAIAPI_MODEL"],
    temperature=0,
    messages=[
        {"role": "system", "content": "只用证据卡。列支持、反证、未知项和引用ID；不要给下单指令。"},
        {"role": "user", "content": json.dumps(candidate_cards, ensure_ascii=False)},
    ],
)
review = reply.choices[0].message.content or ""
assert review.strip(), "empty review"

六个阶段分别定义通过条件

这套流程把选股拆成可独立验收的阶段。每一步都要留下输入、输出和失败证据，避免最后只剩一个无法复核的股票名单。

回测必须使用当时真实可见的信息

股票池要包含历史成分变化与退市样本；财报字段按公告发布日期进入；价格数据明确复权口径；交易模拟包含停牌、涨跌停、手续费和滑点。规则在训练窗口确定后，测试窗口不能继续调参。 比较基准不仅看收益，还看回撤、换手、行业暴露、容量和不同市场阶段。模型版本、Prompt、数据版本和随机性配置都写入实验记录。

每日调度、成本和失败恢复

一次任务成本 = 输入 token × 输入单价 + 输出 token × 输出单价 + 缓存读写费用 + 外部数据费用。模型价格和实际 token 只从当前模型目录与控制台调用日志读取，不在文章里复制一份会过时的数字。 确定性过滤不消耗模型 token，应尽量把全市场缩到可人工复核的候选量。数据抓取按交易日和来源缓存；模型只处理新增证据。401 检查 Key，429 限制并发；公告源失败则暂停对应候选；模型失败可以重试复核，但不得跳过模型而自动进入模拟清单。任务用 trading_date 加 strategy_version 保持幂等。

常见问题