当WorkBuddy遇上本地股票数据：我是如何用ig50搭建专属股票AI助理的

写在前面

我不是什么AI大佬，就是一个普普通通的A股交易者，平时喜欢折腾点新东西。

之前用WorkBuddy做股票分析的时候，最大的感受就是——数据永远是瓶颈。每次问一个问题，WorkBuddy要调API、等返回、解析数据，等结果出来，我当时的判断已经变了。更头疼的是，很多API有频率限制，做批量分析的时候动不动就被限流。而且你把持仓、策略思路都喂给了第三方API，心里总是不踏实——那些写在代码里的买卖逻辑，说穿了都是真金白银砸出来的。

后来我开始用ig50本地数据落盘方案，把整个股票市场的数据搬到自己服务器上，才意识到一个问题：WorkBuddy的能力边界，很大程度上取决于数据的获取方式。当数据从"远程API调用"变成"本地文件读取"，整个分析流程的效率提升了一个量级。

今天想分享的，就是WorkBuddy如何结合本地股票数据ig50，构建一套真正好用的专属股票AI助理。

---

第一阶段：从"问一句答一句"到"自主分析"

最初的痛点

一开始用WorkBuddy做股票分析，基本是这样的流程：

我: "帮我看看今天主力资金流入最多的10只股票"
WorkBuddy: [调用API] → [等待返回] → [解析数据] → "以下是今天主力资金流入最多的..."
我: "再看看这几只股票的龙虎榜情况"
WorkBuddy: [调用API] → [等待返回] → [解析数据] → "以下是龙虎榜数据..."
我: "再分析一下这几只股票的量价关系"
WorkBuddy: [调用API] → [等待返回] → [解析数据] → "以下是量价分析..."

这样用的问题是：每次都是被动的问答，WorkBuddy没有自主分析的能力。而且API调用有延迟，有频率限制，做多维度分析的时候效率很低。

更烦人的是，API返回的数据经常不完整。比如问主力资金流向，有的股票数据有，有的没有；问历史K线，复权价格有时候对不上。这种数据质量，你敢拿来做实盘决策吗？

第一个转折：ig50本地数据解放了WorkBuddy

ig50的安装其实很简单，Linux下一行命令就能搞定（详细的安装指引可以参考官方文档：start-guide.html），Windows下也是下个压缩包解压就行。装好之后，程序会24小时自动更新数据，不用你管。

换成本地数据方案之后，最大的变化是——WorkBuddy不再需要"等待API返回"了。

数据在本地文件里：
/ig50-data/time/a/real/time/{code}      # A股实时行情（3秒落盘）
/ig50-data/time/a/history/trade/{code}   # A股历史K线（日/周/月/季/年）
/ig50-data/time/zijin/                     # 主力资金流向
/ig50-data/time/zdtgc/                      # 涨跌停股池（每10分钟更新）
/ig50-data/time/rzrq/                       # 融资融券数据
/ig50-data/base/gplist                       # 股票列表
...

WorkBuddy要做分析，直接读本地文件就行：

import json
import pandas as pd
from pathlib import Path

DATA_DIR = Path('/ig50-data')

# WorkBuddy直接读取本地数据
def load_realtime_quote(code):
    """读取单只股票实时行情"""
    path = DATA_DIR / 'time' / 'a' / 'real' / 'time' / code
    with open(path, 'r', encoding='utf-8') as f:
        data = json.load(f)
    return data

# 读取平安银行的实时行情
quote = load_realtime_quote('000001')
print(f"价格: {quote.get('p')}, 涨跌幅: {quote.get('pc')}%")

没有网络延迟，没有频率限制，想查多少次就查多少次。

更重要的是，WorkBuddy可以自主跑完整分析流程了：

我: "帮我分析一下今天的市场概况"
WorkBuddy: 
  → 读取沪深京主要指数行情（本地，毫秒级）
  → 读取两市成交额数据（本地，毫秒级）
  → 读取涨跌停股池（本地，毫秒级）
  → 读取主力资金流向（本地，毫秒级）
  → 读取北向资金数据（本地，毫秒级）
  → 融合分析，输出市场概况
  → "截至14:38，上证指数3128.45（+0.82%），深证成指9856.72（+1.24%）..."

从"我问一句它答一句"到"我给个目标它自主跑完"，这是质的提升。

---

第二阶段：ig50数据全景——WorkBuddy能读到什么

221个数据集，覆盖全市场

ig50提供的数据集非常全，一共221个，覆盖沪深京A股、港股、美股、基金市场。我整理了一下WorkBuddy最常用的几个数据源：

数据类别	数据集数量	本地路径前缀	更新频率	典型用途
股票列表	3个	/base/gplist	每天16点	获取股票代码池
实时行情	7个	/time/{market}/real/time	3秒落盘	实时监控、预警
历史K线	19个级别	/time/{market}/history/trade	每日更新	技术分析、回测
资金流向	6个	/time/zijin/	每天20:00	主力动向分析
涨跌停股池	5个	/time/zdtgc/	每10分钟	情绪温度监控
融资融券	4个	/time/rzrq/	每天08:50	杠杆情绪监控
F10基本面	20个+	/time/f10/	每天15:30	基本面排雷
龙虎榜	2个	/time/lhb/	盘后更新	机构席位追踪

数据全是标准JSON格式，字段一目了然。比如实时行情的数据结构：

{
  "code": "000001",
  "name": "平安银行",
  "p": 12.35,
  "pc": 2.35,
  "o": 12.10,
  "h": 12.45,
  "l": 12.00,
  "v": 85623400,
  "a": 1056234000
}

字段名虽然简洁，但含义很明确：p是当前价，pc是涨跌幅，o是开盘，h是最高，l是最低，v是成交量，a是成交额。WorkBuddy读这些数据完全不用做复杂的格式转换。

多线程读取，速度取决于硬件

更关键的是，读本地的速度取决于你的CPU、内存和硬盘速度。如果用多线程并行读取，配合SSD和多核CPU，性能还能进一步提升。

from concurrent.futures import ThreadPoolExecutor

def load_data_file(file_path):
    """WorkBuddy多线程读取单个本地数据文件"""
    with open(file_path, 'r', encoding='utf-8') as f:
        data = json.load(f)
    return data

def batch_load_quotes(codes, max_workers=8):
    """批量读取多只股票的实时行情"""
    paths = [DATA_DIR / 'time' / 'a' / 'real' / 'time' / code for code in codes]
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        results = list(executor.map(load_data_file, paths))
    
    return results

# 同时读取100只股票的实时行情
codes = get_all_stock_codes()[:100]
quotes = batch_load_quotes(codes)

实际测试下来的速度对比：

股票数量	API调用耗时	本地单线程读取	本地多线程读取
10只	约5秒	约10毫秒	约5毫秒
50只	约25秒	约50毫秒	约15毫秒
100只	约50秒	约100毫秒	约25毫秒
500只	不可行	约500毫秒	约100毫秒

效率提升了几百倍，这就是本地数据的威力。

---

第三阶段：构建自然语言问答体系

WorkBuddy最擅长的：自然语言 → 数据查询

有了本地数据，WorkBuddy最强大的能力就被释放出来了——把自然语言转换成数据查询逻辑。

我给WorkBuddy写了一个系统prompt，告诉它数据目录结构和字段含义：

SYSTEM_PROMPT = """
你是一个专业的股票数据分析助手。你可以直接读取本地数据文件来回答用户问题。

数据目录结构：
- /ig50-data/base/gplist/  → 股票列表，字段：code（代码）、name（名称）、exchange（交易所）
- /ig50-data/time/a/real/time/{code}  → A股实时行情，字段：p（价格）、pc（涨跌幅%）、v（成交量）
- /ig50-data/time/a/history/trade/{code}/day  → 日K线，字段：date、open、high、low、close、volume
- /ig50-data/time/zijin/zjlrqs/{code}  → 资金流入趋势，字段：date、主力净流入（元）
- /ig50-data/time/zdtgc/ztgc/{date}  → 涨停股池，字段：code、name、涨幅%、封板时间

查询规则：
1. 用户问"今天行情怎么样"，你需要：
   - 读取主要指数行情（sh000001, sz399001, sz399006）
   - 读取当天涨跌停股池
   - 读取北向资金数据
   - 综合输出市场概况
2. 用户问"某只股票怎么样"，你需要：
   - 读取该股票实时行情
   - 读取最近10天K线
   - 读取最近资金流向
   - 综合分析后输出
3. 所有数据都来自本地文件，不要调用任何外部API
"""

有了这个prompt之后，我就可以用自然语言直接问了：

我: "今天行情怎么样？"
WorkBuddy:
  → 读取上证指数、深证成指、创业板指的实时行情（本地）
  → 读取今天的涨停股池、跌停股池（本地）
  → 读取主力资金流向排名（本地）
  → 读取北向资金数据（本地）
  → 综合输出：
    "截至14:38，沪深京主要指数表现：
     上证指数 3128.45（+0.82%）
     深证成指 9856.72（+1.24%）
     创业板指 1923.18（+1.87%）
     两市成交额 8642 亿元，较昨日放量 12.3%
     上涨 3284 家，下跌 1867 家
     涨停 58 家，跌停 7 家
     主力资金净流入前三：半导体（+48.6亿）、AI算力（+32.4亿）、创新药（+21.8亿）
     北向资金净买入 42.6 亿元，已连续 5 日净流入"

整个过程全在本地完成，数据毫秒级读取，不用等任何API返回。

更多问答场景

除了市场概况，我还让WorkBuddy支持了这些常见查询：

用户问题	WorkBuddy读取的数据源	输出内容
“贵州茅台最近怎么样？”	实时行情+近10天K线+资金流向	价格走势、资金动向、技术面分析
“今天有哪些涨停股？”	涨停股池（ztgc）	涨停股票列表、连板数、封板资金排名
“主力资金流入最多的板块是哪个？”	行业资金流向	板块主力净流入排名、领涨领跌股
“帮我找出连续3天主力净流入的股票”	近3天资金流向（zjlrqs）	筛选后的股票列表、净流入金额排序
“最近一周北向资金买了哪些股？”	北向资金数据	北向增持排名、增持股数、增持金额

这些查询如果用API做，要么要调很多次，要么根本做不到。但用本地数据，WorkBuddy可以自由组合任意数据源，想怎么查就怎么查。

---

第四阶段：实时监控与自动预警

WorkBuddy做监控的优势：可以7x24小时跑

之前用API做监控，最大的问题是API调用有频率限制，做不到真正的实时。换成本地数据之后，WorkBuddy可以每隔几秒就扫一遍数据，真正的实时监控。

import time
from datetime import datetime

def real_time_monitor(watch_list, interval=3):
    """WorkBuddy实时监控核心股票池"""
    
    while True:
        now = datetime.now()
        
        # 只在交易时段运行
        if not is_trading_time(now):
            time.sleep(60)
            continue
        
        # WorkBuddy读取实时行情数据（本地，毫秒级）
        for code in watch_list:
            realtime_path = DATA_DIR / 'time' / 'a' / 'real' / 'time' / code
            with open(realtime_path, 'r', encoding='utf-8') as f:
                rt_data = json.load(f)
            
            price = float(rt_data.get('p', 0))
            change_pct = float(rt_data.get('pc', 0))
            volume = float(rt_data.get('v', 0))
            
            # WorkBuddy判断是否触发预警
            if change_pct > 5:
                send_alert(f"⚠️ {code} 涨幅超过5%，当前涨幅{change_pct:.2f}%")
            
            if volume > rt_data.get('avg_volume', 0) * 3:
                send_alert(f"📊 {code} 成交量异常放大，是平均的{volume/rt_data['avg_volume']:.1f}倍")
        
        # 每隔几秒扫描一次
        time.sleep(interval)

def send_alert(message):
    """WorkBuddy发送预警消息"""
    print(f"[{datetime.now().strftime('%H:%M:%S')}] {message}")
    # 实际应用中可以接入企业微信、钉钉、飞书等通知渠道

这套监控体系的核心优势是：

1. 真正的实时：每3秒扫描一次，不是等API返回的"伪实时"
2. 无频率限制：本地数据，想扫多少次扫多少次
3. 可扩展性强：监控池从10只扩展到100只，性能几乎无影响

6种预警场景

我把常用的预警场景固化成6种：

预警类型	数据来源	触发条件	WorkBuddy行为
涨幅预警	实时行情	涨幅>5%	立刻推送
跌幅预警	实时行情	跌幅>3%	立刻推送
量能异常	实时行情	成交量>3倍平均	推送分析
主力异动	资金流向	主力流入>1000万	推送详情
涨停预警	涨停股池	进入涨停池	推送提醒
炸板预警	涨停股池	涨停股打开	推送警告

WorkBuddy会持续扫描这些条件，一旦触发就立刻推送。这对于短线交易来说，价值巨大——不需要一直盯着屏幕，WorkBuddy会帮你盯着。

全市场异动扫描

除了监控自选股，WorkBuddy还可以扫描整个市场的异动：

def full_market_anomaly_scan():
    """WorkBuddy全市场异动扫描"""
    
    # 获取所有A股代码
    all_codes = get_all_stock_codes()
    
    # 多线程批量读取实时行情
    quotes = batch_load_quotes(all_codes, max_workers=16)
    
    anomalies = {
        'rocket_launch': [],    # 火箭发射（涨速超2%）
        'high_dive': [],        # 高台跳水（跌速超2%）
        'volume_boom': [],      # 放量（成交量>3倍平均）
        'limit_up': [],         # 接近涨停（涨幅>9%）
        'limit_down': []        # 接近跌停（跌幅<-9%）
    }
    
    for q in quotes:
        code = q.get('code')
        change_pct = float(q.get('pc', 0))
        volume_ratio = float(q.get('v', 0)) / max(float(q.get('avg_volume', 1)), 1)
        
        if change_pct > 9:
            anomalies['limit_up'].append((code, change_pct))
        elif change_pct < -9:
            anomalies['limit_down'].append((code, change_pct))
        elif change_pct > 2:
            anomalies['rocket_launch'].append((code, change_pct))
        elif change_pct < -2:
            anomalies['high_dive'].append((code, change_pct))
        
        if volume_ratio > 3:
            anomalies['volume_boom'].append((code, volume_ratio))
    
    return anomalies

# 每分钟扫描一次全市场
while True:
    anomalies = full_market_anomaly_scan()
    if anomalies['limit_up']:
        send_alert(f"🔥 涨停股：{len(anomalies['limit_up'])}只")
    if anomalies['rocket_launch']:
        send_alert(f"🚀 快速拉升：{len(anomalies['rocket_launch'])}只")
    time.sleep(60)

5500多只A股，WorkBuddy一分钟内就能扫描完，找出所有异动股。

---

第五阶段：策略回测与代码生成

大白话 → 策略代码 → 回测验证

这是我觉得最神奇的功能——用大白话描述一个策略，WorkBuddy会自己去读历史K线数据，写出回测代码，然后跑回测给你看结果。

我: "我想测试一个均线金叉策略：5日均线上穿20日均线买入，下穿卖出，过去三年在沪深300成分股上的表现如何？"

WorkBuddy:
  → 理解策略逻辑：均线金叉/死叉
  → 读取沪深300成分股列表（本地）
  → 读取每只股票过去3年的日K线数据（本地）
  → 编写回测代码
  → 运行回测
  → 输出结果

生成的回测代码大概是这样的：

import numpy as np

def ma_crossover_strategy(df, short_window=5, long_window=20):
    """WorkBuddy生成的均线金叉策略"""
    
    df = df.copy()
    df['ma_short'] = df['close'].rolling(window=short_window).mean()
    df['ma_long'] = df['close'].rolling(window=long_window).mean()
    
    # 金叉买入信号
    df['signal'] = 0
    df.loc[df['ma_short'] > df['ma_long'], 'signal'] = 1
    df.loc[df['ma_short'] <= df['ma_long'], 'signal'] = 0
    
    # 计算收益
    df['returns'] = df['close'].pct_change()
    df['strategy_returns'] = df['signal'].shift(1) * df['returns']
    
    # 累计收益
    df['cumulative_returns'] = (1 + df['strategy_returns']).cumprod()
    
    return df

def backtest_on_index(index_code, start_date, end_date):
    """WorkBuddy在指数成分股上回测"""
    
    # 获取成分股列表
    stocks = get_index_components(index_code)
    
    results = []
    
    for code in stocks:
        # 读取历史K线（本地，毫秒级）
        kline_path = DATA_DIR / 'time' / 'a' / 'history' / 'trade' / code / 'day'
        with open(kline_path, 'r', encoding='utf-8') as f:
            kline_data = json.load(f)
        
        df = pd.DataFrame(kline_data)
        df['date'] = pd.to_datetime(df['date'])
        df = df[(df['date'] >= start_date) & (df['date'] <= end_date)]
        
        if len(df) < 60:
            continue
        
        # 运行策略
        df = ma_crossover_strategy(df)
        
        # 计算绩效指标
        total_return = df['cumulative_returns'].iloc[-1] - 1
        win_rate = (df['strategy_returns'] > 0).sum() / max((df['strategy_returns'] != 0).sum(), 1)
        max_drawdown = (df['cumulative_returns'] / df['cumulative_returns'].cummax() - 1).min()
        
        results.append({
            'code': code,
            'total_return': total_return,
            'win_rate': win_rate,
            'max_drawdown': max_drawdown
        })
    
    return pd.DataFrame(results)

# 运行回测
result = backtest_on_index('000300', '2022-01-01', '2024-12-31')
print(f"平均收益: {result['total_return'].mean():.2%}")
print(f"平均胜率: {result['win_rate'].mean():.2%}")
print(f"平均最大回撤: {result['max_drawdown'].mean():.2%}")

然后WorkBuddy会把回测结果报给你：

均线金叉策略在沪深300成分股上的回测结果（2022-01-01 至 2024-12-31）：

📊 平均收益：+8.5%
🎯 平均胜率：52.3%
📉 平均最大回撤：-15.2%

表现最好的10只股票：
1. 600519 贵州茅台：+45.2%，胜率58.3%，最大回撤-12.1%
2. 000858 五粮液：+38.7%，胜率56.7%，最大回撤-14.5%
...

表现最差的10只股票：
1. 601318 中国平安：-22.3%，胜率47.2%，最大回撤-35.6%
...

整个过程——从理解你的自然语言描述，到写代码，到跑回测，到输出分析报告——全是WorkBuddy自主完成的。

多策略对比回测

你还可以让WorkBuddy同时跑多个策略做对比：

strategies = {
    '均线金叉': ma_crossover_strategy,
    '布林带突破': bollinger_breakout_strategy,
    'MACD金叉': macd_crossover_strategy,
    '动量选股': momentum_strategy
}

# WorkBuddy同时跑4个策略
comparison = compare_strategies(strategies, '000300', '2022-01-01', '2024-12-31')

输出的对比表格：

策略	平均收益	平均胜率	平均最大回撤	夏普比率
布林带突破	+12.5%	55.3%	-10.2%	0.85
均线金叉	+8.5%	52.3%	-15.2%	0.52
MACD金叉	+6.8%	51.7%	-13.8%	0.45
动量选股	+5.2%	49.8%	-18.5%	0.31

---

第六阶段：踩坑与优化

教训一：数据更新时间差

本地数据虽然快，但也要注意更新时间。不同数据集的更新时间不一样：

数据集	更新时间	注意事项
实时行情	交易时段每3秒	最及时的数据
涨跌停股池	交易时段每10分钟	盘中持续更新
主力资金流向	每天20:00	盘后更新，盘中不可用
F10数据	每天15:30	盘后更新，次日可用
融资融券	每天08:50	次日上午更新前一天的数据

WorkBuddy会自动判断数据的新鲜度：

import os

def check_data_freshness(data_path):
    """WorkBuddy检查数据新鲜度"""
    mod_time = os.path.getmtime(data_path)
    now = time.time()
    
    hours_old = (now - mod_time) / 3600
    
    if hours_old > 24:
        return f"⚠️ 数据已超过24小时未更新"
    elif hours_old > 1:
        return f"📊 数据更新于{hours_old:.1f}小时前"
    else:
        return f"✅ 数据最近1小时内已更新"

教训二：不能完全依赖WorkBuddy

WorkBuddy可以帮你快速获取数据、分析、回测，但最终买卖决策还是得自己来。框架是工具，不是圣杯。

我的原则是：WorkBuddy给我候选池，我做人脑决策。WorkBuddy的分析结果只是参考，我还会结合自己的经验、盘感做最后的判断。

比如WorkBuddy推送了一个"主力资金大幅流入"的预警，但我一看这只股票是ST股，而且基本面很差，那我肯定不会买。工具可以帮你提高效率，但不能替你做判断。

教训三：数据格式要注意

ig50的数据虽然是标准JSON，但不同数据集的字段命名可能略有差异。比如有的地方用"p"表示价格，有的地方用"close"；有的用"pc"表示涨跌幅，有的用"change_pct"。

我让WorkBuddy写了一个数据标准化模块：

def standardize_quote_data(data):
    """WorkBuddy统一行情数据格式"""
    field_mapping = {
        'p': 'price',
        'pc': 'change_pct',
        'o': 'open',
        'h': 'high',
        'l': 'low',
        'v': 'volume',
        'a': 'amount'
    }
    
    result = {}
    for old_key, new_key in field_mapping.items():
        if old_key in data:
            result[new_key] = data[old_key]
        else:
            result[new_key] = data.get(new_key)
    
    return result

---

总结：WorkBuddy+本地数据的正确姿势

用WorkBuddy结合ig50本地数据做股票分析的这段时间，我的整个研究方式彻底变了：

之前：API时代

• 问一个问题要等几秒到几十秒
• 频率限制，批量分析做不了
• 数据不完整，质量参差不齐
• 策略和持仓要传给第三方

现在：本地数据时代

• 毫秒级数据读取，想问就问
• 没有频率限制，想扫就扫
• 221个数据集，全覆盖高质量
• 所有数据在自己服务器上，隐私安全

从"等API返回"到"毫秒级读取"，从"被动问答"到"自主分析"，效率提升了一个量级。

---

给同路人的建议

如果你也想用WorkBuddy结合ig50本地数据搭建自己的股票AI助理，建议从这几步开始：

1. 先装ig50跑起来：按照安装指引一步一步来，Linux下一行命令就行，先让数据跑起来。最低配置8核16G，推荐用云服务器，24小时不关机。

2. 确定你的核心需求：你是要做实时监控？还是要做策略回测？还是要做基本面分析？不同的需求，用到的数据集不一样。

3. 从简单的问答开始：不要一开始就搞复杂的系统。先让WorkBuddy能回答"今天行情怎么样""某只股票最近如何"这种简单问题，然后再逐步扩展。

4. 数据安全第一：虽然数据在本地，但也要做好备份。ig50的数据量不小，建议定期备份关键数据。

把这些问题想清楚，再去构建你的WorkBuddy分析体系。ig50提供2周免费试用，先试试，觉得适合自己再继续。数据是放大器，你的方法论才是核心。

---

以上内容仅供交流学习，不构成投资建议。股市有风险，入市需谨慎。