我用StockApi如何根据活跃游资选股,附python代码

用户头像sh_**711i03
2026-11-01 发布

游资活跃股共振:别人画像,你直接看聚合结果

本文由 StockApi 智能体基于最近游资活跃股、月游资排行榜与龙虎榜接口整理撰写,代码可直接运行复现。

引言

跟游资的难点从来不是"找不到游资",而是画像成本太高:拉三个月交割单、按席位统计胜率、再盯当日龙虎榜交叉验证——这套流程认真做完要一天,等你做完,行情走完了。

但有一类"懒人友好"的数据:平台已经替你完成聚合的游资信号。比如「最近五日各大游资买入金额大于 8000 万且关注度比较高的票」这种现成股池——它已经把"哪些票正在被多个大资金营业部反复光顾"这件事算好了。你要做的不是从零画像,而是把聚合股池、月度席位排行、当日龙虎榜三层共振起来用:股池给"最近谁热",月榜给"哪路游资在主导",龙虎榜给"今天有没有真金白银确认"。

策略原理

本篇用到StockApi的四个接口

接口(中文名) 路径 在策略里做什么
最近游资活跃股 /v1/activeYouziStocks 近 5 日游资大额买入 + 高关注度的聚合股池(10 次/天,必须落库,交易日 18:00 更新)
月游资排行榜单 /v1/youziRank 当月席位买入排行,识别主导资金风格
龙虎榜查询 /v1/base/dragonTiger 当日上榜明细与净买额,做最后确认
股票日/周/月 K 线行情 /v1/base/day 近 5 日涨幅,过滤已处高位的热度股(涨幅 > 60% 只看不买)

字段中文对照

字段 中文含义 备注
total_count 总上榜次数 近 5 日该票被游资关注的总次数
big_count 大额营业部次数 单笔 > 8000 万的营业部上榜次数
yyb_count 营业部个数 涉及的营业部数量,越分散越健康
diff_total_count 较昨日总次数变化 增量信号:今天又多了一批游资进来
diff_yyb_count 较昨日营业部个数变化 新营业部加入 = 关注度扩散
total_je_formatted 总净额(格式化字符串) 已格式化,数值比较需清洗
bk_info 所属板块 用于板块聚集度判断
close(日 K 口径) 个股收盘价 5 日涨幅 = 收盘 / 5 日前收盘 − 1,过滤高位票
buyAmountRatio 买入额占成交额比(龙虎榜口径) ≥ 5% 说明上榜买入是真仓位

三层共振流程(18:00 数据更新后)

  1. 热度层(游资活跃股接口):取 total_count ≥ 3 且 big_count ≥ 2 的票;重点看增量:diff_total_count > 0 或 diff_yyb_count > 0 的票——游资是做加法的票才有接力价值,次数开始环比下降的票是热度尾声。
  2. 主导层(月游资排行接口):确认当月排行前列的席位属于"接力型"还是"一日游型"——接力型席位主导的活跃股,次日参与胜率显著更高;这个判断不需要精算,看月榜前十的名字性质即可。
  3. 确认层(龙虎榜接口):候选票当日若上龙虎榜,要求买入额占成交额比 ≥ 5% 且净买额为正——把"热度高但今日实际在出货"的票剔除。
  4. 位置过滤(日 K 接口):近 5 日涨幅 > 60% 的票只观察不参与——游资活跃股里最危险的就是这些高位票,热度还在但赔率已经反转。
  5. 执行:通过全部过滤的票,次日开盘涨幅 ≤ 3% 参与,持有 3 个交易日,跌破买入价 5% 离场。

与"游资席位跟随"篇(前作)的分工:那一篇是从交割单自下而上画像(适合深研少数席位);这一篇是自上而下用聚合结果(适合每天批量过一遍全市场)。前者精度高、成本高;后者成本低、覆盖广。两条路都通,按精力分配。

为什么"增量字段"比"存量字段"值钱

活跃股池里 total_count 大的票,可能已经热了四天——你看到的时候就是接盘的时候。真正有预测力的是 diff_* 系列变化量:

状态 total_count diff_total_count 含义 动作
热度启动 3—5 +1 或更多 新游资刚进场 最佳介入窗口
热度维持 6—10 ≈ 0 老面孔原地踏步 观察,等增量
热度衰竭 ≥ 8 负数 游资开始撤 回避

聚合接口最容易被用成"买热度最高的票"——正好用反。存量表筛选,增量表决策,这是所有聚合类数据的通用法则。

Python 完整实现

import os
import re
import time
import requests
import pandas as pd

TOKEN = os.getenv("STOCKAPI_TOKEN", "你的Token")
BASE = "//www.stockapi.com.cn/v1"

def call(path, **params):
    params["token"] = TOKEN
    r = requests.get(f"{BASE}{path}", params=params, timeout=15)
    r.raise_for_status()
    d = r.json()
    if d.get("code") == 20000:
        return d["data"]
    raise RuntimeError(f"{path} 调用失败: {d.get('msg')}")

def to_num(s):
    """清洗 'total_je_formatted' 这类格式化字符串(如 '1.2亿'、'3456万')"""
    if isinstance(s, (int, float)):
        return float(s)
    m = re.match(r"([\d.]+)(亿|万)?", str(s))
    if not m:
        return None
    v = float(m.group(1))
    return v * 1e8 if m.group(2) == "亿" else v * 1e4 if m.group(2) == "万" else v

# ---------- 1. 热度层:活跃股池(每天最多 10 次调用,落库复用) ----------
def active_pool(trade_date):
    df = pd.DataFrame(call("/activeYouziStocks", tradeDate=trade_date) or [])
    if df.empty:
        return df
    for c in ["total_count", "big_count", "yyb_count",
              "diff_total_count", "diff_yyb_count"]:
        if c in df.columns:
            df[c] = pd.to_numeric(df[c], errors="coerce")
    if "total_je_formatted" in df.columns:
        df["total_je"] = df["total_je_formatted"].map(to_num)
    df["code"] = df["code"].astype(str).str.zfill(6)
    return df

# ---------- 2. 日 K:5 日涨幅过滤高位票 ----------
def zf5(code):
    d = call("/base/day", code=code, startDate="2026-09-01",
             endDate="2026-10-29", calculationCycle=100)
    k = pd.DataFrame({c: v for c, v in d.items() if isinstance(v, list)})
    k["time"] = pd.to_datetime(k["time"])
    k = k.sort_values("time")
    k["close"] = pd.to_numeric(k["close"], errors="coerce")
    if len(k) < 6:
        return None
    return k["close"].iloc[-1] / k["close"].iloc[-6] - 1

# ---------- 3. 确认层:龙虎榜净买 ----------
def dragon_ok(code, trade_date):
    df = pd.DataFrame(call("/base/dragonTiger", tradeDate=trade_date) or [])
    if df.empty:
        return False
    for c in df.columns:
        try:
            df[c] = pd.to_numeric(df[c], errors="ignore")
        except (TypeError, ValueError):
            pass
    row = df[df["code"].astype(str).str.zfill(6) == code]
    if row.empty:
        return True   # 未上榜不否决,交给热度层判断
    return row["buyAmountRatio"].iloc[0] >= 5

# ---------- 4. 主流程:增量共振 ----------
def resonance(date):
    pool = active_pool(date)
    if pool.empty:
        return pd.DataFrame()
    hot = pool[(pool["total_count"] >= 3) & (pool["big_count"] >= 2) &
               ((pool["diff_total_count"] > 0) | (pool["diff_yyb_count"] > 0))]
    picks = []
    for c in hot["code"]:
        z = zf5(c)          # 高位票过滤:5 日涨幅 > 60% 只观察
        if z is not None and z > 0.60:
            continue
        if dragon_ok(c, date):
            picks.append(c)
    out = hot[hot["code"].isin(picks)].copy()
    return out.sort_values("diff_total_count", ascending=False)

res = resonance("2026-10-28")
print(res[["code", "name", "total_count", "diff_total_count",
           "big_count"]].head(10).to_string(index=False))

回测指标怎么读

指标 典型量级(示意) 说明
每日共振票数 1—6 只 增量条件过滤后并不多
3 日持有胜率 55%—65% 龙虎榜净买确认贡献主要
平均持有收益 2%—6% 3 日短持有,快进快出
最大风险 高位票断板补跌 用开盘 ≤ 3% 与 5% 止损控制

活跃股池每天只有 10 次调用,回测需逐日落库积累;先用龙虎榜 + 涨停池近似回放,落库两周后再用真实聚合数据校准。

参数与风控

参数 默认 调整方向
总上榜次数下限 3 提高到 5 更聚焦头部
增量条件 diff 任一 > 0 收紧为 diff_total_count ≥ 2
次日参与上限 开盘 ≤ 3% 高开太多的票次日溢价已被吃掉
持有期 3 交易日 游资接力周期通常不超过一周

风控三条:

  1. 接口 10 次/天的调用限额是硬约束:每天收盘后调用一次并落库,绝不在盘中浪费调用次数。
  2. 候选票若位于 5 日涨幅 > 60% 的位置,只观察不参与——游资活跃股里高危的正是这些高位票。
  3. 月榜席位的风格会漂移(接力型游资也会改打一日游),每月重新评估一次主导席位性质,不要用上个月的印象。

常见问题(FAQ)

Q:活跃股池和交割单画像(游资席位跟随篇)数据有什么区别?
A:活跃股池是平台预聚合的股票维度数据(哪些票热、增量如何),交割单是原始的席位维度数据(谁在买、买多少)。前者开箱即用但无法细拆席位;后者自由度最高但要自己写画像逻辑。按"批量筛选用池、深度研究用单"的原则分配。

Q:total_je_formatted 为什么是字符串?
A:该字段是给人看的格式化文本("1.2亿")。做数值比较必须先清洗:正则提取数字与单位(亿/万),换算回元。代码里的 to_num 就是干这个的——直接 pd.to_numeric 会全部变 NaN。

Q:为什么候选票没上龙虎榜也算通过?
A:龙虎榜有上榜门槛,游资小仓位试错的票不一定上榜。未上榜不否决,上榜才做净买校验——龙虎榜在这一层是"证伪工具"(发现上榜但净卖出的票),不是"证实工具"。

评论