当WorkBuddy遇上本地数据我是如何用本地数据引擎构建智能选股体系的写在前面我不是什么量化大佬就是一个普普通通的A股交易者。之前用WorkBuddy做股票分析的时候最大的感受就是——数据调用的速度永远跟不上思路的变化。每次问一个问题WorkBuddy要去调API、等返回、解析数据等结果出来我当时的判断已经变了。更头疼的是很多API有频率限制做批量分析的时候动不动就被限流。后来我开始用本地数据落盘方案把数据从远程API落盘到本地文件才意识到一个问题WorkBuddy的能力边界很大程度上取决于数据的获取方式。当数据从远程API调用变成本地文件读取整个分析流程的效率提升了一个量级。今天想分享的就是WorkBuddy如何结合本地股票数据ig50构建一套真正好用的智能选股体系。第一阶段从问WorkBuddy到WorkBuddy主动跑最初的痛点一开始用WorkBuddy做股票分析基本是这样的流程我: 帮我看看今天主力资金流入最多的10只股票 WorkBuddy: [调用API] → [等待返回] → [解析数据] → 以下是今天主力资金流入最多的... 我: 再看看这几只股票的龙虎榜情况 WorkBuddy: [调用API] → [等待返回] → [解析数据] → 以下是龙虎榜数据... 我: 再分析一下这几只股票的量价关系 WorkBuddy: [调用API] → [等待返回] → [解析数据] → 以下是量价分析...这样用的问题是每次都是被动的问答WorkBuddy没有自主分析的能力。而且API调用有延迟有频率限制做多维度分析的时候效率很低。第一个转折本地数据解放了WorkBuddy的能力换成本地数据方案之后最大的变化是——WorkBuddy不再需要等待API返回了。数据在本地文件里 /data/stock-a/L2行情指标3秒落盘 /data/stock-a/主力资金流向 /data/stock-a/龙虎榜 /data/stock-a/当天逐笔交易 ...WorkBuddy要做分析直接读本地文件就行importjsonimportpandasaspdfrompathlibimportPath DATA_DIRPath(/data)# WorkBuddy直接读取本地数据defload_main_fund_flow():读取主力资金流向数据pathDATA_DIR/stock-a/主力资金流向withopen(path,r,encodingutf-8)asf:datajson.load(f)returnpd.DataFrame(data)# 一行代码毫秒级返回dfload_main_fund_flow()top_inflowdf.sort_values(主力净流入(万元),ascendingFalse).head(10)没有网络延迟没有频率限制想查多少次就查多少次。更重要的是WorkBuddy可以自主跑完整分析流程了我: 帮我选出今天强势股 WorkBuddy: → 读取主力资金流向数据本地多线程并行读取秒级完成 → 读取涨停股池数据本地多线程并行读取秒级完成 → 读取北向资金数据本地多线程并行读取秒级完成 → 读取龙虎榜数据本地多线程并行读取秒级完成 → 融合分析输出候选股池 → 以下是我筛选的强势股按照主力资金流入涨停强度北向增持综合排名...读本地的速度取决于你的CPU、内存和硬盘速度——SSD多核CPU充足内存几百只股票的数据几秒钟就能读完。从我问一句它答一句到我给个目标它自主跑完这是质的提升。第二阶段构建多维选股框架WorkBuddy的优势可以同时处理多个数据源用API的时候做多维分析很痛苦——每个维度的数据要单独调用还要等API返回最后手动拼接。用本地数据之后WorkBuddy可以同时读取多个数据源多线程并行处理秒级完成融合分析。这个速度的上限取决于本地CPU、内存和硬盘速度。我跟WorkBuddy构建了这样一套选股框架fromconcurrent.futuresimportThreadPoolExecutordefload_data_file(file_path):WorkBuddy多线程读取单个本地数据文件withopen(file_path,r,encodingutf-8)asf:datajson.load(f)returnpd.DataFrame(data)defsmart_stock_selection():智能选股融合多个数据源# WorkBuddy用多线程并行读取多个本地数据文件data_files[DATA_DIR/stock-a/主力资金流向,DATA_DIR/stock-a/涨停股池,DATA_DIR/stock-center/北向资金,DATA_DIR/stock-a/龙虎榜]withThreadPoolExecutor(max_workers4)asexecutor:fund_flow,limit_up,north_money,block_tradelist(executor.map(load_data_file,data_files))# 融合分析逻辑# 维度一主力资金大幅流入1000万元strong_fundfund_flow[fund_flow[主力净流入(万元)]1000]# 维度二涨停或接近涨停涨幅9%strong_pricelimit_up[limit_up[涨幅(%)]9]# 维度三北向资金增持north_buynorth_money[north_money[增持股数(万股)]10]# 维度四大单交易活跃active_tradeblock_trade[block_trade[成交金额(万元)]500]# WorkBuddy自动计算交集找出同时满足多个条件的股票candidatesset(strong_fund[股票代码])set(strong_price[股票代码])# 输出候选股池result[]forcodeincandidates:score0ifcodeinstrong_fund[股票代码].values:score30ifcodeinnorth_buy[股票代码].values:score25ifcodeinactive_trade[股票代码].values:score20result.append({code:code,score:score,reason:主力资金流入涨停北向增持大单活跃})returnsorted(result,keylambdax:x[score],reverseTrue)这套框架的核心是WorkBuddy不用等我一步步问它直接把所有数据源读完给我一个综合评分的结果。五维强势股模型经过反复调整我们固化了一套五维强势股评分模型维度数据来源筛选条件权重资金流向主力资金流向主力净流入1000万30分价格强势涨停股池涨幅9%25分北向增持北向资金增持股数10万股20分机构动向龙虎榜机构净买入500万15分成交活跃大单交易成交金额500万10分WorkBuddy每天会自动跑一遍这个模型输出一个按评分排序的候选股池。我要做的就是在这个候选池里做最后的决策。第三阶段实时监控与预警WorkBuddy做监控的优势可以7x24小时跑之前用API做监控最大的问题是API调用有频率限制做不到真正的实时。换成本地数据之后WorkBuddy可以每隔几秒就扫一遍数据真正的实时监控。importtimefromdatetimeimportdatetimedefreal_time_monitor():WorkBuddy实时监控核心股票池watch_list[000001,000002,600000,600036]# 监控池whileTrue:nowdatetime.now()# WorkBuddy读取实时行情数据本地毫秒级forcodeinwatch_list:realtime_pathDATA_DIR/time/real/codewithopen(realtime_path,r,encodingutf-8)asf:rt_datajson.load(f)pricefloat(rt_data.get(p,0))change_pctfloat(rt_data.get(pc,0))volumefloat(rt_data.get(v,0))# WorkBuddy判断是否触发预警ifchange_pct5:alert(f⚠️{code}涨幅超过5%当前涨幅{change_pct:.2f}%)ifvolumert_data.get(avg_volume,0)*3:alert(f{code}成交量异常放大当前成交量是平均的{volume/rt_data[avg_volume]:.1f}倍)# 每3秒扫描一次time.sleep(3)defalert(message):WorkBuddy发送预警消息print(f[{datetime.now().strftime(%H:%M:%S)}]{message})# 实际应用中可以接入企业微信、钉钉等通知渠道这套监控体系的核心优势是真正的实时每3秒扫描一次不是等API返回的伪实时无频率限制本地数据想扫多少次扫多少次可扩展性强监控池从10只扩展到100只性能几乎无影响6种预警场景我们把预警场景固化成6种预警类型触发条件WorkBuddy行为涨幅预警涨幅5%推送消息跌幅预警跌幅3%推送消息量能异常成交量3倍平均推送分析主力异动主力资金大幅流入/流出推送详情涨停预警股价接近涨停推送提醒炸板预警涨停股打开涨停推送警告WorkBuddy会持续扫描这些条件一旦触发就立刻推送。这对于短线交易来说价值巨大——不需要一直盯着屏幕WorkBuddy会帮你盯着。第四阶段历史数据回测与策略优化WorkBuddy做回测的优势可以批量跑多个策略之前做策略回测最头疼的是数据获取太慢。跑一个策略要先从API拉历史数据拉一周的分钟线就要等好几秒。如果要跑多个策略、多个时间段时间成本很高。换成本地数据之后WorkBuddy可以批量读取历史数据快速完成回测defbacktest_strategy(strategy_func,start_date,end_date):WorkBuddy回测策略results[]# WorkBuddy读取历史K线数据本地毫秒级kline_pathDATA_DIR/stock-a/历史K线forstock_codeinget_all_stock_codes():# 读取单只股票的历史K线stock_klineread_kline_data(kline_path,stock_code,start_date,end_date)ifstock_klineisNone:continue# WorkBuddy运行策略计算收益signalsstrategy_func(stock_kline)# 计算策略收益profitcalculate_profit(signals,stock_kline)results.append({stock:stock_code,profit:profit,win_rate:calculate_win_rate(signals,stock_kline),max_drawdown:calculate_max_drawdown(signals,stock_kline)})returnpd.DataFrame(results)# WorkBuddy批量跑多个策略strategy_results[]forstrategyin[ma_strategy,bollinger_strategy,momentum_strategy]:resultbacktest_strategy(strategy,2024-01-01,2024-12-31)result[strategy]strategy.__name__ strategy_results.append(result)# 对比不同策略的表现comparisonpd.concat(strategy_results).groupby(strategy).agg({profit:mean,win_rate:mean,max_drawdown:mean})WorkBuddy可以在几分钟内跑完多个策略、多只股票、多个时间段的回测这在用API的时候是不可想象的。回测框架的核心能力我们构建的回测框架有几个核心能力能力描述数据来源分钟级回测支持1分钟、5分钟、15分钟等分钟级回测历史K线分钟级日线级回测支持日线、周线、月线回测历史K线日线以上资金流向回测结合主力资金流向分析策略有效性主力资金流向事件驱动回测结合涨停、龙虎榜等事件分析涨停股池、龙虎榜WorkBuddy可以自主选择回测粒度自主分析策略表现给我一个完整的评估报告。第五阶段踩坑与优化教训一数据格式不统一刚开始用的时候发现不同数据源的字段名称、数据格式不完全一致。比如主力资金流向用的是主力净流入(万元)“而北向资金用的是增持股数(万股)”。后来WorkBuddy帮我写了一套数据标准化模块defstandardize_fund_data(df,value_col):WorkBuddy统一资金数据格式# 统一单位为万元if股数invalue_col:df[value_col]df[value_col]/10# 万股转万元假设股价10元# 统一列名dfdf.rename(columns{value_col:资金净流入(万元)})returndf教训二数据更新时间差本地数据虽然快但也要注意更新时间。比如主力资金流向数据通常是盘后更新的而实时行情数据是盘中秒级更新的。WorkBuddy会自动判断数据的时间戳defcheck_data_freshness(data_path):WorkBuddy检查数据新鲜度importos mod_timeos.path.getmtime(data_path)nowtime.time()hours_old(now-mod_time)/3600ifhours_old24:print(f⚠️ 数据已超过24小时未更新)elifhours_old1:print(f 数据更新于{hours_old:.1f}小时前)else:print(f✅ 数据最近1小时内已更新)教训三不能完全依赖WorkBuddyWorkBuddy可以帮你快速筛选、分析但最终买卖决策还是得自己来。框架是工具不是圣杯。我的原则是WorkBuddy给我候选池我做人脑决策。WorkBuddy的评分只是参考我还会结合自己的经验、盘感做最后的判断。总结WorkBuddy本地数据的正确姿势用WorkBuddy结合本地数据做股票分析的这段时间我总结了几个心得✅ 正确的用法用WorkBuddy快速获取数据本地数据让WorkBuddy可以毫秒级读取没有API延迟用WorkBuddy构建分析框架WorkBuddy可以同时处理多个数据源融合分析用WorkBuddy持续监控预警WorkBuddy可以7x24小时跑监控不遗漏机会❌ 错误的用法期望WorkBuddy给你稳赚不赔的策略完全依赖WorkBuddy做交易决策不理解逻辑就盲目跟单给同路人的建议如果你也想用WorkBuddy结合本地数据做股票分析建议从这几个问题开始你的分析维度有哪些资金/技术面/基本面你需要什么粒度的数据分钟级/日线级你的监控需求是什么实时/盘后把这些问题想清楚再去构建你的WorkBuddy分析体系。本地数据是放大器你的方法论才是核心。以上内容仅供交流学习不构成投资建议。股市有风险入市需谨慎。
当WorkBuddy遇上本地数据我是如何用本地数据引擎构建智能选股体系的写在前面我不是什么量化大佬就是一个普普通通的A股交易者。之前用WorkBuddy做股票分析的时候最大的感受就是——数据调用的速度永远跟不上思路的变化。每次问一个问题WorkBuddy要去调API、等返回、解析数据等结果出来我当时的判断已经变了。更头疼的是很多API有频率限制做批量分析的时候动不动就被限流。后来我开始用本地数据落盘方案把数据从远程API落盘到本地文件才意识到一个问题WorkBuddy的能力边界很大程度上取决于数据的获取方式。当数据从远程API调用变成本地文件读取整个分析流程的效率提升了一个量级。今天想分享的就是WorkBuddy如何结合本地股票数据ig50构建一套真正好用的智能选股体系。第一阶段从问WorkBuddy到WorkBuddy主动跑最初的痛点一开始用WorkBuddy做股票分析基本是这样的流程我: 帮我看看今天主力资金流入最多的10只股票 WorkBuddy: [调用API] → [等待返回] → [解析数据] → 以下是今天主力资金流入最多的... 我: 再看看这几只股票的龙虎榜情况 WorkBuddy: [调用API] → [等待返回] → [解析数据] → 以下是龙虎榜数据... 我: 再分析一下这几只股票的量价关系 WorkBuddy: [调用API] → [等待返回] → [解析数据] → 以下是量价分析...这样用的问题是每次都是被动的问答WorkBuddy没有自主分析的能力。而且API调用有延迟有频率限制做多维度分析的时候效率很低。第一个转折本地数据解放了WorkBuddy的能力换成本地数据方案之后最大的变化是——WorkBuddy不再需要等待API返回了。数据在本地文件里 /data/stock-a/L2行情指标3秒落盘 /data/stock-a/主力资金流向 /data/stock-a/龙虎榜 /data/stock-a/当天逐笔交易 ...WorkBuddy要做分析直接读本地文件就行importjsonimportpandasaspdfrompathlibimportPath DATA_DIRPath(/data)# WorkBuddy直接读取本地数据defload_main_fund_flow():读取主力资金流向数据pathDATA_DIR/stock-a/主力资金流向withopen(path,r,encodingutf-8)asf:datajson.load(f)returnpd.DataFrame(data)# 一行代码毫秒级返回dfload_main_fund_flow()top_inflowdf.sort_values(主力净流入(万元),ascendingFalse).head(10)没有网络延迟没有频率限制想查多少次就查多少次。更重要的是WorkBuddy可以自主跑完整分析流程了我: 帮我选出今天强势股 WorkBuddy: → 读取主力资金流向数据本地多线程并行读取秒级完成 → 读取涨停股池数据本地多线程并行读取秒级完成 → 读取北向资金数据本地多线程并行读取秒级完成 → 读取龙虎榜数据本地多线程并行读取秒级完成 → 融合分析输出候选股池 → 以下是我筛选的强势股按照主力资金流入涨停强度北向增持综合排名...读本地的速度取决于你的CPU、内存和硬盘速度——SSD多核CPU充足内存几百只股票的数据几秒钟就能读完。从我问一句它答一句到我给个目标它自主跑完这是质的提升。第二阶段构建多维选股框架WorkBuddy的优势可以同时处理多个数据源用API的时候做多维分析很痛苦——每个维度的数据要单独调用还要等API返回最后手动拼接。用本地数据之后WorkBuddy可以同时读取多个数据源多线程并行处理秒级完成融合分析。这个速度的上限取决于本地CPU、内存和硬盘速度。我跟WorkBuddy构建了这样一套选股框架fromconcurrent.futuresimportThreadPoolExecutordefload_data_file(file_path):WorkBuddy多线程读取单个本地数据文件withopen(file_path,r,encodingutf-8)asf:datajson.load(f)returnpd.DataFrame(data)defsmart_stock_selection():智能选股融合多个数据源# WorkBuddy用多线程并行读取多个本地数据文件data_files[DATA_DIR/stock-a/主力资金流向,DATA_DIR/stock-a/涨停股池,DATA_DIR/stock-center/北向资金,DATA_DIR/stock-a/龙虎榜]withThreadPoolExecutor(max_workers4)asexecutor:fund_flow,limit_up,north_money,block_tradelist(executor.map(load_data_file,data_files))# 融合分析逻辑# 维度一主力资金大幅流入1000万元strong_fundfund_flow[fund_flow[主力净流入(万元)]1000]# 维度二涨停或接近涨停涨幅9%strong_pricelimit_up[limit_up[涨幅(%)]9]# 维度三北向资金增持north_buynorth_money[north_money[增持股数(万股)]10]# 维度四大单交易活跃active_tradeblock_trade[block_trade[成交金额(万元)]500]# WorkBuddy自动计算交集找出同时满足多个条件的股票candidatesset(strong_fund[股票代码])set(strong_price[股票代码])# 输出候选股池result[]forcodeincandidates:score0ifcodeinstrong_fund[股票代码].values:score30ifcodeinnorth_buy[股票代码].values:score25ifcodeinactive_trade[股票代码].values:score20result.append({code:code,score:score,reason:主力资金流入涨停北向增持大单活跃})returnsorted(result,keylambdax:x[score],reverseTrue)这套框架的核心是WorkBuddy不用等我一步步问它直接把所有数据源读完给我一个综合评分的结果。五维强势股模型经过反复调整我们固化了一套五维强势股评分模型维度数据来源筛选条件权重资金流向主力资金流向主力净流入1000万30分价格强势涨停股池涨幅9%25分北向增持北向资金增持股数10万股20分机构动向龙虎榜机构净买入500万15分成交活跃大单交易成交金额500万10分WorkBuddy每天会自动跑一遍这个模型输出一个按评分排序的候选股池。我要做的就是在这个候选池里做最后的决策。第三阶段实时监控与预警WorkBuddy做监控的优势可以7x24小时跑之前用API做监控最大的问题是API调用有频率限制做不到真正的实时。换成本地数据之后WorkBuddy可以每隔几秒就扫一遍数据真正的实时监控。importtimefromdatetimeimportdatetimedefreal_time_monitor():WorkBuddy实时监控核心股票池watch_list[000001,000002,600000,600036]# 监控池whileTrue:nowdatetime.now()# WorkBuddy读取实时行情数据本地毫秒级forcodeinwatch_list:realtime_pathDATA_DIR/time/real/codewithopen(realtime_path,r,encodingutf-8)asf:rt_datajson.load(f)pricefloat(rt_data.get(p,0))change_pctfloat(rt_data.get(pc,0))volumefloat(rt_data.get(v,0))# WorkBuddy判断是否触发预警ifchange_pct5:alert(f⚠️{code}涨幅超过5%当前涨幅{change_pct:.2f}%)ifvolumert_data.get(avg_volume,0)*3:alert(f{code}成交量异常放大当前成交量是平均的{volume/rt_data[avg_volume]:.1f}倍)# 每3秒扫描一次time.sleep(3)defalert(message):WorkBuddy发送预警消息print(f[{datetime.now().strftime(%H:%M:%S)}]{message})# 实际应用中可以接入企业微信、钉钉等通知渠道这套监控体系的核心优势是真正的实时每3秒扫描一次不是等API返回的伪实时无频率限制本地数据想扫多少次扫多少次可扩展性强监控池从10只扩展到100只性能几乎无影响6种预警场景我们把预警场景固化成6种预警类型触发条件WorkBuddy行为涨幅预警涨幅5%推送消息跌幅预警跌幅3%推送消息量能异常成交量3倍平均推送分析主力异动主力资金大幅流入/流出推送详情涨停预警股价接近涨停推送提醒炸板预警涨停股打开涨停推送警告WorkBuddy会持续扫描这些条件一旦触发就立刻推送。这对于短线交易来说价值巨大——不需要一直盯着屏幕WorkBuddy会帮你盯着。第四阶段历史数据回测与策略优化WorkBuddy做回测的优势可以批量跑多个策略之前做策略回测最头疼的是数据获取太慢。跑一个策略要先从API拉历史数据拉一周的分钟线就要等好几秒。如果要跑多个策略、多个时间段时间成本很高。换成本地数据之后WorkBuddy可以批量读取历史数据快速完成回测defbacktest_strategy(strategy_func,start_date,end_date):WorkBuddy回测策略results[]# WorkBuddy读取历史K线数据本地毫秒级kline_pathDATA_DIR/stock-a/历史K线forstock_codeinget_all_stock_codes():# 读取单只股票的历史K线stock_klineread_kline_data(kline_path,stock_code,start_date,end_date)ifstock_klineisNone:continue# WorkBuddy运行策略计算收益signalsstrategy_func(stock_kline)# 计算策略收益profitcalculate_profit(signals,stock_kline)results.append({stock:stock_code,profit:profit,win_rate:calculate_win_rate(signals,stock_kline),max_drawdown:calculate_max_drawdown(signals,stock_kline)})returnpd.DataFrame(results)# WorkBuddy批量跑多个策略strategy_results[]forstrategyin[ma_strategy,bollinger_strategy,momentum_strategy]:resultbacktest_strategy(strategy,2024-01-01,2024-12-31)result[strategy]strategy.__name__ strategy_results.append(result)# 对比不同策略的表现comparisonpd.concat(strategy_results).groupby(strategy).agg({profit:mean,win_rate:mean,max_drawdown:mean})WorkBuddy可以在几分钟内跑完多个策略、多只股票、多个时间段的回测这在用API的时候是不可想象的。回测框架的核心能力我们构建的回测框架有几个核心能力能力描述数据来源分钟级回测支持1分钟、5分钟、15分钟等分钟级回测历史K线分钟级日线级回测支持日线、周线、月线回测历史K线日线以上资金流向回测结合主力资金流向分析策略有效性主力资金流向事件驱动回测结合涨停、龙虎榜等事件分析涨停股池、龙虎榜WorkBuddy可以自主选择回测粒度自主分析策略表现给我一个完整的评估报告。第五阶段踩坑与优化教训一数据格式不统一刚开始用的时候发现不同数据源的字段名称、数据格式不完全一致。比如主力资金流向用的是主力净流入(万元)“而北向资金用的是增持股数(万股)”。后来WorkBuddy帮我写了一套数据标准化模块defstandardize_fund_data(df,value_col):WorkBuddy统一资金数据格式# 统一单位为万元if股数invalue_col:df[value_col]df[value_col]/10# 万股转万元假设股价10元# 统一列名dfdf.rename(columns{value_col:资金净流入(万元)})returndf教训二数据更新时间差本地数据虽然快但也要注意更新时间。比如主力资金流向数据通常是盘后更新的而实时行情数据是盘中秒级更新的。WorkBuddy会自动判断数据的时间戳defcheck_data_freshness(data_path):WorkBuddy检查数据新鲜度importos mod_timeos.path.getmtime(data_path)nowtime.time()hours_old(now-mod_time)/3600ifhours_old24:print(f⚠️ 数据已超过24小时未更新)elifhours_old1:print(f 数据更新于{hours_old:.1f}小时前)else:print(f✅ 数据最近1小时内已更新)教训三不能完全依赖WorkBuddyWorkBuddy可以帮你快速筛选、分析但最终买卖决策还是得自己来。框架是工具不是圣杯。我的原则是WorkBuddy给我候选池我做人脑决策。WorkBuddy的评分只是参考我还会结合自己的经验、盘感做最后的判断。总结WorkBuddy本地数据的正确姿势用WorkBuddy结合本地数据做股票分析的这段时间我总结了几个心得✅ 正确的用法用WorkBuddy快速获取数据本地数据让WorkBuddy可以毫秒级读取没有API延迟用WorkBuddy构建分析框架WorkBuddy可以同时处理多个数据源融合分析用WorkBuddy持续监控预警WorkBuddy可以7x24小时跑监控不遗漏机会❌ 错误的用法期望WorkBuddy给你稳赚不赔的策略完全依赖WorkBuddy做交易决策不理解逻辑就盲目跟单给同路人的建议如果你也想用WorkBuddy结合本地数据做股票分析建议从这几个问题开始你的分析维度有哪些资金/技术面/基本面你需要什么粒度的数据分钟级/日线级你的监控需求是什么实时/盘后把这些问题想清楚再去构建你的WorkBuddy分析体系。本地数据是放大器你的方法论才是核心。以上内容仅供交流学习不构成投资建议。股市有风险入市需谨慎。
