1. 从“感觉”到“数据”量化买卖压力的实战价值在A股市场里泡久了很多朋友都有过类似的体验感觉某只股票“抛压很重”或者“买盘很强”但真要你拿出个标准来往往又说不清道不明。盘口上密密麻麻的挂单分时图上忽上忽下的成交这些信息背后到底藏着多少真实的买卖意愿传统的技术指标比如成交量、换手率虽然能反映活跃度但很难精准区分是主动买入推动的上涨还是被动卖出导致的放量。而“量化买卖压力”这个思路就是试图用一套可计算、可回溯的规则把这种市场感觉数据化、模型化。简单来说量化买卖压力选股的核心是试图通过分析订单簿和逐笔成交数据来量化一段时间内市场多空双方的真实力量对比。它不是看K线的形态也不是猜主力的意图而是直接对最原始的买卖行为进行统计和建模。比如我们可以定义一个指标将每一笔成交按照其成交价与前一档买卖报价的关系区分为“主动性买入”和“主动性卖出”。所有主动性买入的成交金额之和构成了“买压”所有主动性卖出成交金额之和构成了“卖压”。两者的差值或比值就构成了我们衡量股票短期资金流向和情绪强弱的依据。这套方法的价值在于它的客观性和可复制性。情绪会骗人但数据不会。通过程序化的方式我们可以快速扫描全市场几千只股票找出当前买卖压力出现显著背离如买压持续大于卖压但股价尚未大幅上涨的标的作为潜在的观察或交易对象。这为从海量信息中高效筛选目标提供了一条清晰的路径。接下来我将结合一个完整的Python实现案例拆解构建一个买卖压力量化选股策略的每一个环节包括数据获取、核心指标计算、策略逻辑设计以及完整的代码实现。你会发现将模糊的盘感转化为严谨的量化信号并没有想象中那么复杂。2. 策略基石买卖压力指标的定义与计算逻辑构建任何量化模型的第一步都是明确并量化你的核心思想。对于买卖压力我们需要一个严格、无歧义的计算定义。这里介绍一种基于Level-2逐笔成交数据的主流计算方法它比普通成交量包含更多信息。2.1 理解逐笔成交与订单簿A股的Level-2数据中每一笔成交都会明确记录其成交价格、成交量以及成交时间精确到毫秒。更重要的是我们可以结合同一时刻的订单簿快照买一、卖一价及挂单量来判断这笔成交的性质。关键逻辑在于如果一笔成交的成交价等于卖一价那么这笔成交的买方是主动的他“吃掉了”卖一档的挂单我们将其记为主动性买入Buy Volume。反之如果成交价等于买一价那么卖方是主动的他“砸向了”买一档的挂单记为主动性卖出Sell Volume。如果成交价同时等于买一和卖一价这种情况在流动性好的股票中很少见或者由于数据切片原因无法判断则通常将其归类为中性成交或忽略。2.2 计算买卖压力比BSPR基于上述逻辑我们可以为每只股票在特定周期例如每分钟、每日内计算买卖压力指标。一个常用且有效的指标是买卖压力比Buy Sell Pressure Ratio, BSPR。它的计算步骤如下数据准备获取目标股票在计算周期内的所有逐笔成交数据以及对应时间点的订单簿快照至少需要买一价bid1和卖一价ask1。逐笔分类遍历每一笔成交。若成交价 ask1则主动性买入金额 成交价 * 成交量若成交价 bid1则主动性卖出金额 成交价 * 成交量其他情况暂不归类或根据更复杂的规则处理。周期汇总将一个周期如一天内所有主动性买入金额累加得到周期总买压Total_Buy_Pressure同样累加得到周期总卖压Total_Sell_Pressure。计算比率买卖压力比BSPR Total_Buy_Pressure / Total_Sell_Pressure。这个指标的意义是什么BSPR 1表明在该周期内主动买入的资金力量超过了主动卖出的力量市场内部需求旺盛。BSPR 1表明主动卖出的力量占优内部抛压较大。BSPR的值越大说明买压相对于卖压越强。单纯的放量上涨BSPR可能接近1和买盘主动推动的上涨BSPR显著大于1其市场含义是不同的。注意在实际计算中直接使用金额还是成交量股数存在细微差别。使用金额更能真实反映资金流向因为它考虑了价格权重。例如高价股上的一笔大单和低价股上的一笔大单即使股数相同其代表的资金量和市场影响力也不同。因此在下面的代码实现中我们将采用成交金额进行计算。2.3 指标的现实考量与优化点在实际应用中直接使用原始的BSPR可能会遇到一些问题极端值对于交易非常不活跃的股票可能一天只有几笔成交计算出的BSPR容易产生极端值如除数为0导致无穷大。因此通常需要设置一个最低成交金额门槛过滤掉流动性过差的股票。趋势与波动单日的BSPR波动可能很大。为了平滑噪声并捕捉趋势我们可以使用移动平均线例如计算5日或10日的BSPR均值。市场状态调整在普涨牛市中多数股票的BSPR可能都大于1在普跌熊市中则相反。因此有时使用BSPR的行业排名或全市场百分位比使用绝对值更有意义。有了清晰的定义我们就可以进入实战环节看看如何用代码获取数据并实现这些计算。3. 实战环境搭建与数据获取方案量化策略的筋骨是逻辑血肉则是数据。没有可靠、高质量的数据再精妙的策略也是空中楼阁。这里我将介绍一个基于Python的、相对可行且低成本的数据获取与处理方案。3.1 核心工具栈选择我们选择Python作为实现语言因为它拥有极其丰富的量化分析库。数据处理与分析pandas和numpy是基石用于处理表格数据和高性能计算。数据获取这里是一个难点。理想的Level-2逐笔数据通常需要从专业数据商购买价格不菲。对于个人研究者和初学者我们可以采用一种折中但有效的替代方案使用akshare库。akshare提供了丰富的免费金融数据接口虽然其“逐笔成交”数据并非官方Level-2而是基于行情推送的整理但对于理解原理和构建初级策略模型来说其包含的成交价格、成交量以及买卖方向通过type字段区分信息已经足够我们模拟计算买卖压力。这完全符合“一名合格从业者在资源有限情况下最可能采用的合理方案”。可视化使用matplotlib或plotly进行结果可视化辅助分析。回测框架可选如果想进行历史回测可以使用backtrader、zipline或自建回测引擎。本文侧重于策略核心计算与选股逻辑暂不展开复杂回测。首先确保环境已安装必要库pip install pandas numpy akshare matplotlib3.2 利用AkShare获取模拟买卖压力数据akshare的stock_zh_a_tick_tx函数可以获取指定股票近期的分笔成交数据通常为最近几个交易日。每笔数据包含时间、价格、成交量、成交额以及一个关键的type字段例如‘买盘’ ‘卖盘’ ‘中性盘’。虽然这不是基于订单簿的精确分类但数据源本身已根据成交驱动方做了初步划分我们可以直接利用这个type字段来近似计算买卖压力。这是一种非常实用的变通方法。下面是一个获取数据并初步处理的函数示例import akshare as ak import pandas as pd def get_tick_data(stock_code, trade_date): 获取指定股票和日期的分笔成交数据 :param stock_code: 股票代码带交易所前缀如 sz000001 :param trade_date: 交易日期格式 20230815 :return: 包含买卖方向标记的DataFrame # 注意akshare的接口可能调整这里以当前可用接口为例 # stock_zh_a_tick_tx 可能需要日期参数且数据是当日的 try: # 尝试获取数据这里假设接口返回包含[时间,价格,成交量,成交额,type]的DataFrame df_tick ak.stock_zh_a_tick_tx(symbolstock_code, trade_datetrade_date) # 确保列名正确并选择所需列 # 实际列名可能需要根据akshare返回结果调整例如可能是‘成交时间’、‘成交价’、‘成交量(手)’等 df_tick df_tick.rename(columns{ 成交时间: time, 成交价: price, 成交量: volume, 成交额: amount, 类型: type # 这个字段标记买卖方向 }) # 计算每笔成交的金额如果原始数据没有 if amount not in df_tick.columns: df_tick[amount] df_tick[price] * df_tick[volume] * 100 # 假设volume是手数需乘以100 # 将时间字符串转换为datetime格式便于后续按分钟或小时聚合 df_tick[datetime] pd.to_datetime(trade_date df_tick[time]) df_tick.set_index(datetime, inplaceTrue) return df_tick[[price, volume, amount, type]] except Exception as e: print(f获取{stock_code}在{trade_date}的数据失败: {e}) return pd.DataFrame() # 示例获取平安银行某日数据 df get_tick_data(sz000001, 20230815) print(df.head())这个函数返回的DataFrame中type字段就是我们区分买卖压力的关键。通常‘买盘’或‘B’表示主动性买入‘卖盘’或‘S’表示主动性卖出。4. 核心代码实现买卖压力计算与选股逻辑有了数据我们就可以实现核心的计算逻辑并构建一个简单的选股策略。4.1 计算单只股票的日度买卖压力比我们首先实现一个函数用于计算单只股票在单个交易日内的BSPR。def calculate_daily_bspr(tick_df): 根据分笔成交数据计算日度买卖压力比(BSPR) :param tick_df: 包含[price,volume,amount,type]的DataFrame :return: 字典包含买压、卖压、BSPR等指标 if tick_df.empty: return None # 根据type字段分类汇总成交金额 # 注意实际字段值需要根据akshare返回结果确认这里用‘买盘’、‘卖盘’举例 buy_mask tick_df[type].str.contains(买盘|B, naFalse) # 假设包含‘买盘’或‘B’为主动买 sell_mask tick_df[type].str.contains(卖盘|S, naFalse) # 假设包含‘卖盘’或‘S’为主动卖 total_buy_pressure tick_df.loc[buy_mask, amount].sum() total_sell_pressure tick_df.loc[sell_mask, amount].sum() # 避免除零错误 if total_sell_pressure 0: bspr float(inf) # 或定义为一个大数如9999 else: bspr total_buy_pressure / total_sell_pressure # 计算总成交额用于验证 total_amount tick_df[amount].sum() return { date: tick_df.index.date[0], total_buy_pressure: total_buy_pressure, total_sell_pressure: total_sell_pressure, bspr: bspr, total_amount: total_amount, buy_ratio: total_buy_pressure / total_amount if total_amount 0 else 0 }4.2 设计一个简单的选股策略单纯的BSPR大于1并不足以构成买入理由。我们需要一个更稳健的策略逻辑。例如我们可以寻找“买压持续强势且股价尚未大幅反应”的股票。一个简单的策略逻辑可以设计为过滤流动性剔除日均成交额过小的股票例如最近5日日均成交额小于1亿元避免流动性风险和数据噪音。计算指标计算股票最近N个交易日例如5日的每日BSPR。定义信号条件A最近5日的BSPR均值大于一个阈值例如1.2。条件B最近5日内BSPR大于1的天数占多数例如至少4天。条件C股价在过去5日内涨幅不大例如累计涨幅小于5%避免追高。综合判断当股票同时满足条件A、B、C时将其列入候选观察列表。4.3 批量处理与选股程序实现下面我们将上述逻辑整合实现一个批量处理股票池的选股程序。import numpy as np from datetime import datetime, timedelta def screen_stocks_by_pressure(stock_list, start_date, end_date, min_avg_amount1e8): 根据买卖压力策略筛选股票 :param stock_list: 股票代码列表如 [sz000001, sh600000] :param start_date: 开始日期20230801 :param end_date: 结束日期20230815 :param min_avg_amount: 最小日均成交额元默认1亿 :return: 符合条件的股票列表及详细指标 results [] # 生成日期序列 date_range pd.date_range(startstart_date, endend_date, freqB) # 交易日 for stock_code in stock_list: print(f正在处理 {stock_code}...) daily_metrics [] # 1. 获取多日数据并计算每日指标 for single_date in date_range: date_str single_date.strftime(%Y%m%d) tick_df get_tick_data(stock_code, date_str) if not tick_df.empty: daily_metric calculate_daily_bspr(tick_df) if daily_metric: daily_metrics.append(daily_metric) if len(daily_metrics) 5: # 数据不足5天跳过 continue df_metrics pd.DataFrame(daily_metrics).set_index(date).sort_index() # 2. 流动性过滤计算近期日均成交额 recent_days min(5, len(df_metrics)) avg_amount df_metrics[total_amount].tail(recent_days).mean() if avg_amount min_avg_amount: continue # 流动性不达标跳过 # 3. 计算策略条件 # 取最近5个交易日数据 lookback_days 5 recent_data df_metrics.tail(lookback_days) if len(recent_data) lookback_days: continue # 条件ABSPR均值 阈值 mean_bspr recent_data[bspr].replace([float(inf), -float(inf)], np.nan).mean() # 条件BBSPR 1的天数占比 days_bspr_gt1 (recent_data[bspr] 1).sum() # 条件C股价涨幅这里需要额外获取价格数据简化处理假设我们已有一个收盘价序列 close_prices # 此处为演示我们用一个假设函数 get_close_price 来获取收盘价 # close_prices get_close_price(stock_code, recent_data.index) # 需要自行实现或从其他接口获取 # price_change (close_prices.iloc[-1] - close_prices.iloc[0]) / close_prices.iloc[0] # 由于篇幅我们暂时注释掉条件C的具体计算假设已计算得到 price_change # 假设我们计算得到的 price_change 为 0.03 (3%) price_change 0.03 # 此处应为实际计算值 # 设置阈值 threshold_mean_bspr 1.2 threshold_days_ratio 0.8 # 80%的天数 BSPR1 threshold_price_change 0.05 # 5% condition_a mean_bspr threshold_mean_bspr condition_b (days_bspr_gt1 / lookback_days) threshold_days_ratio condition_c price_change threshold_price_change if condition_a and condition_b and condition_c: stock_result { stock_code: stock_code, avg_amount(亿): round(avg_amount / 1e8, 2), mean_bspr: round(mean_bspr, 3), days_bspr_gt1: f{days_bspr_gt1}/{lookback_days}, price_change(%): round(price_change * 100, 2), recent_dates: recent_data.index.strftime(%Y-%m-%d).tolist() } results.append(stock_result) # 将结果转换为DataFrame并排序 if results: results_df pd.DataFrame(results) # 可以按 mean_bspr 降序排列买压最强的排前面 results_df results_df.sort_values(bymean_bspr, ascendingFalse) return results_df else: return pd.DataFrame() # 示例运行选股策略 if __name__ __main__: # 一个小的测试股票池 test_pool [sz000001, sz000002, sh600036, sh600519] start 20230807 end 20230811 selected_stocks screen_stocks_by_pressure(test_pool, start, end) print(\n 买卖压力选股策略筛选结果 ) print(selected_stocks)这段代码实现了一个完整的选股流程。它遍历股票池获取历史数据计算买卖压力指标并应用我们预设的三重过滤条件最终输出符合条件的股票列表及其关键指标。5. 策略的深入思考、潜在问题与优化方向将代码跑通只是第一步更重要的是理解其局限性和优化空间。在实际应用中以下几个问题必须考虑。5.1 数据质量与分类准确性我们依赖akshare的type字段进行分类这本质上是一种“推定”的买卖方向。与基于精确订单簿的快照判断判断成交价与ask1/bid1的关系相比可能存在一定误差。例如大单拆分成小单成交、冰山订单等复杂情况可能无法被准确归类。这是使用免费或中低频数据无法避免的折衷。对于严肃的实盘策略投资于高质量的Level-2逐笔委托数据是必要的。5.2 策略的逻辑漏洞与市场环境适应性我们的示例策略逻辑相对简单存在几个明显漏洞未考虑市场整体情绪在单边上涨的牛市中大部分股票BSPR都可能持续大于1此时策略会选出大量股票但其中很多可能是跟随大盘上涨而非独立的强势买盘。解决方法之一是做横截面比较即计算所有股票BSPR的行业排名或全市场百分位只选择排名在前10%或20%的股票。未结合价格趋势我们仅限制了短期涨幅但未考虑中长期趋势。一只处于长期下降通道中的股票即使出现几天买压占优也可能只是下跌中的反弹。可以考虑引入均线系统如股价位于20日或60日均线之上作为趋势过滤条件。未处理极端行情在涨停或跌停板上买卖压力指标会失真。涨停时只有买盘卖压为0BSPR为无穷大跌停时则相反。需要在计算前剔除涨跌停日的数据或对这类日子的指标进行特殊处理如赋予一个固定极值。5.3 从选股到交易的鸿沟选出股票并不意味着立即买入。量化选股模型输出的是一个“观察列表”或“候选池”还需要经过更细致的评估才能进入交易环节人工复核查看候选股票的基本面有无重大变化如业绩暴雷、监管处罚、消息面有无驱动如行业政策、公司公告。技术形态确认结合K线形态、支撑压力位、其他技术指标如MACD、RSI进行二次确认。仓位与风控决定投资比例、设置止损止盈位。一个完整的策略必须包含明确的出场规则例如当BSPR连续两日小于0.8时平仓或股价从买入点回撤超过8%时止损。交易成本模型中需考虑佣金和印花税对于高频或短线策略交易成本的影响尤为显著。5.4 性能优化与扩展思路如果股票池很大如全市场近5000只股票循环获取数据并计算会成为性能瓶颈。优化方法包括向量化计算尽可能使用pandas和numpy的向量化操作避免在Python层级的循环。并行处理使用multiprocessing或concurrent.futures库进行多进程/多线程计算充分利用多核CPU。数据本地化将常用的历史数据提前下载到本地数据库如SQLite、DuckDB或文件中避免每次运行都从网络接口重复获取。指标扩展买卖压力可以衍生出更多指标例如大单净流入只统计单笔成交金额超过一定阈值如50万元的主动性买卖差额。压力趋势计算BSPR的N日移动平均线观察其金叉死叉。压力背离股价创新低但BSPR指标未创新低底背离可能预示下跌动能衰竭。6. 一个完整的策略回测框架搭建思路为了科学地评估买卖压力策略的有效性我们需要进行历史回测。这里给出一个简易回测框架的搭建思路你可以基于此进行扩展。回测的核心是避免“未来函数”即只能用当时已经发生的数据做决策。我们的买卖压力数据是基于当日成交的在实盘中只能在收盘后才能完整计算。因此回测时我们假设在交易日T收盘后根据T日及之前的数据计算出信号在下一个交易日T1的开盘价执行买入/卖出操作。import pandas as pd import numpy as np class SimplePressureBacktester: def __init__(self, initial_capital1000000, commission_rate0.0003): self.initial_capital initial_capital self.commission_rate commission_rate self.positions {} # 持仓字典 {‘code’: shares} self.cash initial_capital self.equity_curve [] # 每日资产总值记录 def run_backtest(self, signal_df, price_df): 运行回测 :param signal_df: 信号DataFrame索引为日期列為股票代码值为1买入/持有或0卖出/不持有 :param price_df: 价格DataFrame索引为日期列為股票代码值为开盘价用于交易 dates signal_df.index.sort_values() prev_signal pd.Series(0, indexsignal_df.columns) # 前一日持仓信号 for date in dates: today_signal signal_df.loc[date] today_price price_df.loc[date] # 1. 处理卖出昨日有持仓但今日信号为0的股票 to_sell prev_signal[prev_signal 1].index.difference(today_signal[today_signal 1].index) for code in to_sell: if code in self.positions and code in today_price: sell_price today_price[code] sell_shares self.positions.pop(code) sell_amount sell_price * sell_shares self.cash sell_amount * (1 - self.commission_rate) # 扣除佣金 # 2. 处理买入今日信号为1但昨日未持仓的股票 # 简易策略等权重分配可用资金给所有新信号股票 to_buy today_signal[today_signal 1].index.difference(prev_signal[prev_signal 1].index) if not to_buy.empty: # 计算每只股票可分配资金扣除预计佣金 available_cash_per_stock self.cash / len(to_buy) for code in to_buy: if code in today_price: buy_price today_price[code] # 计算可买股数以100股为单位 max_shares int((available_cash_per_stock / (1 self.commission_rate)) // (buy_price * 100)) * 100 if max_shares 0: cost buy_price * max_shares * (1 self.commission_rate) self.positions[code] max_shares self.cash - cost # 3. 计算当日总资产 position_value 0 for code, shares in self.positions.items(): if code in today_price: position_value today_price[code] * shares total_equity self.cash position_value self.equity_curve.append({date: date, equity: total_equity}) # 4. 更新前一日信号 prev_signal today_signal.copy() # 回测结束平仓所有头寸假设最后一天以收盘价卖出 last_date dates[-1] last_price price_df.loc[last_date] for code, shares in list(self.positions.items()): if code in last_price: sell_amount last_price[code] * shares self.cash sell_amount * (1 - self.commission_rate) self.positions.clear() final_equity self.cash return pd.DataFrame(self.equity_curve).set_index(date), final_equity # 使用示例 # 1. 需要事先准备好 signal_df 和 price_df # signal_df: 每个交易日每只股票是否有买入/持有信号1/0 # price_df: 每个交易日每只股票的开盘价用于交易 # 2. 初始化回测器并运行 # backtester SimplePressureBacktester(initial_capital1000000) # equity_curve, final_value backtester.run_backtest(signal_df, price_df) # 3. 分析回测结果计算年化收益率、最大回撤、夏普比率等这个框架非常基础忽略了涨跌停无法买卖、滑点等问题但清晰地展示了回测的逻辑流程。在实际应用中你需要根据策略信号生成signal_df并获取对应的历史价格数据price_df。7. 我的实操心得与避坑指南在实践买卖压力策略的过程中我踩过不少坑也积累了一些心得这些是代码和公式里不会写的细节。心得一数据源的稳定性和一致性是生命线。早期我用过一些免费的网页爬虫接口经常因为网站改版而失效导致策略中断。后来我固定使用akshare这类维护相对活跃的库并养成了在策略核心代码外包裹严密异常捕获的习惯。对于关键数据我会定期抽样检查比如对比不同日期同一只股票的数据总量是否在合理范围内防止接口静默返回错误数据。永远不要完全信任你的数据源必须设置数据质量检查点。心得二买卖压力指标不是“圣杯”它揭示的是微观结构下的短期力量对比。我曾一度迷信高BSPR股票但发现很多情况下股价依然不涨。后来我意识到强大的买压可能发生在股价高位是主力对倒出货制造繁荣假象也可能发生在下跌初期是抄底资金与抛压的博弈但最终空头力量更强。因此必须将买卖压力指标放在更大的市场语境中结合大盘趋势、行业板块热度、个股所处的价格位置是突破关键位还是超跌反弹来综合判断。单独使用任何一个指标都是危险的。心得三参数过拟合是量化新手的通病。当我第一次回测时我不断调整BSPR的阈值1.1, 1.2, 1.3...、观察天数3天5天10天...直到在历史数据上获得一条漂亮的净值曲线。但这毫无意义因为这只是在拟合历史噪声。正确的做法是将数据分为样本内和样本外。用样本内数据例如2018-2020年确定一组“逻辑上合理”的参数比如BSPR1.2代表买压显著然后将其应用到完全未参与参数优化的样本外数据例如2021-2022年进行验证。如果策略在样本外依然有效才说明它可能具备一定的稳健性。心得四从想法到实盘风控的权重必须占到80%以上。选股模型只解决了“买什么”的问题而“买多少”、“何时卖”、“错了怎么办”才是决定长期盈亏的关键。在我的实盘框架中买卖压力信号只是一个触发因子它会进入一个包含仓位管理模块、风险平价模块和止损止盈模块的决策系统。例如即使某只股票BSPR信号很强如果当前整体市场风险偏高如波动率指数VIX飙升我的仓位管理模块也会强制降低单只股票的持仓上限。没有风控的量化策略就像没有刹车的跑车。最后我想强调的是本文提供的代码和思路是一个完整的、可运行的起点但它绝非一个“稳赚不赔”的赚钱机器。它的价值在于展示了如何将一种市场直觉感知买卖力量转化为可量化、可检验、可迭代的计算机程序。你可以在此基础上融入你自己的市场理解增加更多的过滤条件结合其他因子如估值、动量、质量等构建属于你自己的复合策略。量化交易的世界里没有一劳永逸的代码只有不断进化的认知和系统。
1. 从“感觉”到“数据”量化买卖压力的实战价值在A股市场里泡久了很多朋友都有过类似的体验感觉某只股票“抛压很重”或者“买盘很强”但真要你拿出个标准来往往又说不清道不明。盘口上密密麻麻的挂单分时图上忽上忽下的成交这些信息背后到底藏着多少真实的买卖意愿传统的技术指标比如成交量、换手率虽然能反映活跃度但很难精准区分是主动买入推动的上涨还是被动卖出导致的放量。而“量化买卖压力”这个思路就是试图用一套可计算、可回溯的规则把这种市场感觉数据化、模型化。简单来说量化买卖压力选股的核心是试图通过分析订单簿和逐笔成交数据来量化一段时间内市场多空双方的真实力量对比。它不是看K线的形态也不是猜主力的意图而是直接对最原始的买卖行为进行统计和建模。比如我们可以定义一个指标将每一笔成交按照其成交价与前一档买卖报价的关系区分为“主动性买入”和“主动性卖出”。所有主动性买入的成交金额之和构成了“买压”所有主动性卖出成交金额之和构成了“卖压”。两者的差值或比值就构成了我们衡量股票短期资金流向和情绪强弱的依据。这套方法的价值在于它的客观性和可复制性。情绪会骗人但数据不会。通过程序化的方式我们可以快速扫描全市场几千只股票找出当前买卖压力出现显著背离如买压持续大于卖压但股价尚未大幅上涨的标的作为潜在的观察或交易对象。这为从海量信息中高效筛选目标提供了一条清晰的路径。接下来我将结合一个完整的Python实现案例拆解构建一个买卖压力量化选股策略的每一个环节包括数据获取、核心指标计算、策略逻辑设计以及完整的代码实现。你会发现将模糊的盘感转化为严谨的量化信号并没有想象中那么复杂。2. 策略基石买卖压力指标的定义与计算逻辑构建任何量化模型的第一步都是明确并量化你的核心思想。对于买卖压力我们需要一个严格、无歧义的计算定义。这里介绍一种基于Level-2逐笔成交数据的主流计算方法它比普通成交量包含更多信息。2.1 理解逐笔成交与订单簿A股的Level-2数据中每一笔成交都会明确记录其成交价格、成交量以及成交时间精确到毫秒。更重要的是我们可以结合同一时刻的订单簿快照买一、卖一价及挂单量来判断这笔成交的性质。关键逻辑在于如果一笔成交的成交价等于卖一价那么这笔成交的买方是主动的他“吃掉了”卖一档的挂单我们将其记为主动性买入Buy Volume。反之如果成交价等于买一价那么卖方是主动的他“砸向了”买一档的挂单记为主动性卖出Sell Volume。如果成交价同时等于买一和卖一价这种情况在流动性好的股票中很少见或者由于数据切片原因无法判断则通常将其归类为中性成交或忽略。2.2 计算买卖压力比BSPR基于上述逻辑我们可以为每只股票在特定周期例如每分钟、每日内计算买卖压力指标。一个常用且有效的指标是买卖压力比Buy Sell Pressure Ratio, BSPR。它的计算步骤如下数据准备获取目标股票在计算周期内的所有逐笔成交数据以及对应时间点的订单簿快照至少需要买一价bid1和卖一价ask1。逐笔分类遍历每一笔成交。若成交价 ask1则主动性买入金额 成交价 * 成交量若成交价 bid1则主动性卖出金额 成交价 * 成交量其他情况暂不归类或根据更复杂的规则处理。周期汇总将一个周期如一天内所有主动性买入金额累加得到周期总买压Total_Buy_Pressure同样累加得到周期总卖压Total_Sell_Pressure。计算比率买卖压力比BSPR Total_Buy_Pressure / Total_Sell_Pressure。这个指标的意义是什么BSPR 1表明在该周期内主动买入的资金力量超过了主动卖出的力量市场内部需求旺盛。BSPR 1表明主动卖出的力量占优内部抛压较大。BSPR的值越大说明买压相对于卖压越强。单纯的放量上涨BSPR可能接近1和买盘主动推动的上涨BSPR显著大于1其市场含义是不同的。注意在实际计算中直接使用金额还是成交量股数存在细微差别。使用金额更能真实反映资金流向因为它考虑了价格权重。例如高价股上的一笔大单和低价股上的一笔大单即使股数相同其代表的资金量和市场影响力也不同。因此在下面的代码实现中我们将采用成交金额进行计算。2.3 指标的现实考量与优化点在实际应用中直接使用原始的BSPR可能会遇到一些问题极端值对于交易非常不活跃的股票可能一天只有几笔成交计算出的BSPR容易产生极端值如除数为0导致无穷大。因此通常需要设置一个最低成交金额门槛过滤掉流动性过差的股票。趋势与波动单日的BSPR波动可能很大。为了平滑噪声并捕捉趋势我们可以使用移动平均线例如计算5日或10日的BSPR均值。市场状态调整在普涨牛市中多数股票的BSPR可能都大于1在普跌熊市中则相反。因此有时使用BSPR的行业排名或全市场百分位比使用绝对值更有意义。有了清晰的定义我们就可以进入实战环节看看如何用代码获取数据并实现这些计算。3. 实战环境搭建与数据获取方案量化策略的筋骨是逻辑血肉则是数据。没有可靠、高质量的数据再精妙的策略也是空中楼阁。这里我将介绍一个基于Python的、相对可行且低成本的数据获取与处理方案。3.1 核心工具栈选择我们选择Python作为实现语言因为它拥有极其丰富的量化分析库。数据处理与分析pandas和numpy是基石用于处理表格数据和高性能计算。数据获取这里是一个难点。理想的Level-2逐笔数据通常需要从专业数据商购买价格不菲。对于个人研究者和初学者我们可以采用一种折中但有效的替代方案使用akshare库。akshare提供了丰富的免费金融数据接口虽然其“逐笔成交”数据并非官方Level-2而是基于行情推送的整理但对于理解原理和构建初级策略模型来说其包含的成交价格、成交量以及买卖方向通过type字段区分信息已经足够我们模拟计算买卖压力。这完全符合“一名合格从业者在资源有限情况下最可能采用的合理方案”。可视化使用matplotlib或plotly进行结果可视化辅助分析。回测框架可选如果想进行历史回测可以使用backtrader、zipline或自建回测引擎。本文侧重于策略核心计算与选股逻辑暂不展开复杂回测。首先确保环境已安装必要库pip install pandas numpy akshare matplotlib3.2 利用AkShare获取模拟买卖压力数据akshare的stock_zh_a_tick_tx函数可以获取指定股票近期的分笔成交数据通常为最近几个交易日。每笔数据包含时间、价格、成交量、成交额以及一个关键的type字段例如‘买盘’ ‘卖盘’ ‘中性盘’。虽然这不是基于订单簿的精确分类但数据源本身已根据成交驱动方做了初步划分我们可以直接利用这个type字段来近似计算买卖压力。这是一种非常实用的变通方法。下面是一个获取数据并初步处理的函数示例import akshare as ak import pandas as pd def get_tick_data(stock_code, trade_date): 获取指定股票和日期的分笔成交数据 :param stock_code: 股票代码带交易所前缀如 sz000001 :param trade_date: 交易日期格式 20230815 :return: 包含买卖方向标记的DataFrame # 注意akshare的接口可能调整这里以当前可用接口为例 # stock_zh_a_tick_tx 可能需要日期参数且数据是当日的 try: # 尝试获取数据这里假设接口返回包含[时间,价格,成交量,成交额,type]的DataFrame df_tick ak.stock_zh_a_tick_tx(symbolstock_code, trade_datetrade_date) # 确保列名正确并选择所需列 # 实际列名可能需要根据akshare返回结果调整例如可能是‘成交时间’、‘成交价’、‘成交量(手)’等 df_tick df_tick.rename(columns{ 成交时间: time, 成交价: price, 成交量: volume, 成交额: amount, 类型: type # 这个字段标记买卖方向 }) # 计算每笔成交的金额如果原始数据没有 if amount not in df_tick.columns: df_tick[amount] df_tick[price] * df_tick[volume] * 100 # 假设volume是手数需乘以100 # 将时间字符串转换为datetime格式便于后续按分钟或小时聚合 df_tick[datetime] pd.to_datetime(trade_date df_tick[time]) df_tick.set_index(datetime, inplaceTrue) return df_tick[[price, volume, amount, type]] except Exception as e: print(f获取{stock_code}在{trade_date}的数据失败: {e}) return pd.DataFrame() # 示例获取平安银行某日数据 df get_tick_data(sz000001, 20230815) print(df.head())这个函数返回的DataFrame中type字段就是我们区分买卖压力的关键。通常‘买盘’或‘B’表示主动性买入‘卖盘’或‘S’表示主动性卖出。4. 核心代码实现买卖压力计算与选股逻辑有了数据我们就可以实现核心的计算逻辑并构建一个简单的选股策略。4.1 计算单只股票的日度买卖压力比我们首先实现一个函数用于计算单只股票在单个交易日内的BSPR。def calculate_daily_bspr(tick_df): 根据分笔成交数据计算日度买卖压力比(BSPR) :param tick_df: 包含[price,volume,amount,type]的DataFrame :return: 字典包含买压、卖压、BSPR等指标 if tick_df.empty: return None # 根据type字段分类汇总成交金额 # 注意实际字段值需要根据akshare返回结果确认这里用‘买盘’、‘卖盘’举例 buy_mask tick_df[type].str.contains(买盘|B, naFalse) # 假设包含‘买盘’或‘B’为主动买 sell_mask tick_df[type].str.contains(卖盘|S, naFalse) # 假设包含‘卖盘’或‘S’为主动卖 total_buy_pressure tick_df.loc[buy_mask, amount].sum() total_sell_pressure tick_df.loc[sell_mask, amount].sum() # 避免除零错误 if total_sell_pressure 0: bspr float(inf) # 或定义为一个大数如9999 else: bspr total_buy_pressure / total_sell_pressure # 计算总成交额用于验证 total_amount tick_df[amount].sum() return { date: tick_df.index.date[0], total_buy_pressure: total_buy_pressure, total_sell_pressure: total_sell_pressure, bspr: bspr, total_amount: total_amount, buy_ratio: total_buy_pressure / total_amount if total_amount 0 else 0 }4.2 设计一个简单的选股策略单纯的BSPR大于1并不足以构成买入理由。我们需要一个更稳健的策略逻辑。例如我们可以寻找“买压持续强势且股价尚未大幅反应”的股票。一个简单的策略逻辑可以设计为过滤流动性剔除日均成交额过小的股票例如最近5日日均成交额小于1亿元避免流动性风险和数据噪音。计算指标计算股票最近N个交易日例如5日的每日BSPR。定义信号条件A最近5日的BSPR均值大于一个阈值例如1.2。条件B最近5日内BSPR大于1的天数占多数例如至少4天。条件C股价在过去5日内涨幅不大例如累计涨幅小于5%避免追高。综合判断当股票同时满足条件A、B、C时将其列入候选观察列表。4.3 批量处理与选股程序实现下面我们将上述逻辑整合实现一个批量处理股票池的选股程序。import numpy as np from datetime import datetime, timedelta def screen_stocks_by_pressure(stock_list, start_date, end_date, min_avg_amount1e8): 根据买卖压力策略筛选股票 :param stock_list: 股票代码列表如 [sz000001, sh600000] :param start_date: 开始日期20230801 :param end_date: 结束日期20230815 :param min_avg_amount: 最小日均成交额元默认1亿 :return: 符合条件的股票列表及详细指标 results [] # 生成日期序列 date_range pd.date_range(startstart_date, endend_date, freqB) # 交易日 for stock_code in stock_list: print(f正在处理 {stock_code}...) daily_metrics [] # 1. 获取多日数据并计算每日指标 for single_date in date_range: date_str single_date.strftime(%Y%m%d) tick_df get_tick_data(stock_code, date_str) if not tick_df.empty: daily_metric calculate_daily_bspr(tick_df) if daily_metric: daily_metrics.append(daily_metric) if len(daily_metrics) 5: # 数据不足5天跳过 continue df_metrics pd.DataFrame(daily_metrics).set_index(date).sort_index() # 2. 流动性过滤计算近期日均成交额 recent_days min(5, len(df_metrics)) avg_amount df_metrics[total_amount].tail(recent_days).mean() if avg_amount min_avg_amount: continue # 流动性不达标跳过 # 3. 计算策略条件 # 取最近5个交易日数据 lookback_days 5 recent_data df_metrics.tail(lookback_days) if len(recent_data) lookback_days: continue # 条件ABSPR均值 阈值 mean_bspr recent_data[bspr].replace([float(inf), -float(inf)], np.nan).mean() # 条件BBSPR 1的天数占比 days_bspr_gt1 (recent_data[bspr] 1).sum() # 条件C股价涨幅这里需要额外获取价格数据简化处理假设我们已有一个收盘价序列 close_prices # 此处为演示我们用一个假设函数 get_close_price 来获取收盘价 # close_prices get_close_price(stock_code, recent_data.index) # 需要自行实现或从其他接口获取 # price_change (close_prices.iloc[-1] - close_prices.iloc[0]) / close_prices.iloc[0] # 由于篇幅我们暂时注释掉条件C的具体计算假设已计算得到 price_change # 假设我们计算得到的 price_change 为 0.03 (3%) price_change 0.03 # 此处应为实际计算值 # 设置阈值 threshold_mean_bspr 1.2 threshold_days_ratio 0.8 # 80%的天数 BSPR1 threshold_price_change 0.05 # 5% condition_a mean_bspr threshold_mean_bspr condition_b (days_bspr_gt1 / lookback_days) threshold_days_ratio condition_c price_change threshold_price_change if condition_a and condition_b and condition_c: stock_result { stock_code: stock_code, avg_amount(亿): round(avg_amount / 1e8, 2), mean_bspr: round(mean_bspr, 3), days_bspr_gt1: f{days_bspr_gt1}/{lookback_days}, price_change(%): round(price_change * 100, 2), recent_dates: recent_data.index.strftime(%Y-%m-%d).tolist() } results.append(stock_result) # 将结果转换为DataFrame并排序 if results: results_df pd.DataFrame(results) # 可以按 mean_bspr 降序排列买压最强的排前面 results_df results_df.sort_values(bymean_bspr, ascendingFalse) return results_df else: return pd.DataFrame() # 示例运行选股策略 if __name__ __main__: # 一个小的测试股票池 test_pool [sz000001, sz000002, sh600036, sh600519] start 20230807 end 20230811 selected_stocks screen_stocks_by_pressure(test_pool, start, end) print(\n 买卖压力选股策略筛选结果 ) print(selected_stocks)这段代码实现了一个完整的选股流程。它遍历股票池获取历史数据计算买卖压力指标并应用我们预设的三重过滤条件最终输出符合条件的股票列表及其关键指标。5. 策略的深入思考、潜在问题与优化方向将代码跑通只是第一步更重要的是理解其局限性和优化空间。在实际应用中以下几个问题必须考虑。5.1 数据质量与分类准确性我们依赖akshare的type字段进行分类这本质上是一种“推定”的买卖方向。与基于精确订单簿的快照判断判断成交价与ask1/bid1的关系相比可能存在一定误差。例如大单拆分成小单成交、冰山订单等复杂情况可能无法被准确归类。这是使用免费或中低频数据无法避免的折衷。对于严肃的实盘策略投资于高质量的Level-2逐笔委托数据是必要的。5.2 策略的逻辑漏洞与市场环境适应性我们的示例策略逻辑相对简单存在几个明显漏洞未考虑市场整体情绪在单边上涨的牛市中大部分股票BSPR都可能持续大于1此时策略会选出大量股票但其中很多可能是跟随大盘上涨而非独立的强势买盘。解决方法之一是做横截面比较即计算所有股票BSPR的行业排名或全市场百分位只选择排名在前10%或20%的股票。未结合价格趋势我们仅限制了短期涨幅但未考虑中长期趋势。一只处于长期下降通道中的股票即使出现几天买压占优也可能只是下跌中的反弹。可以考虑引入均线系统如股价位于20日或60日均线之上作为趋势过滤条件。未处理极端行情在涨停或跌停板上买卖压力指标会失真。涨停时只有买盘卖压为0BSPR为无穷大跌停时则相反。需要在计算前剔除涨跌停日的数据或对这类日子的指标进行特殊处理如赋予一个固定极值。5.3 从选股到交易的鸿沟选出股票并不意味着立即买入。量化选股模型输出的是一个“观察列表”或“候选池”还需要经过更细致的评估才能进入交易环节人工复核查看候选股票的基本面有无重大变化如业绩暴雷、监管处罚、消息面有无驱动如行业政策、公司公告。技术形态确认结合K线形态、支撑压力位、其他技术指标如MACD、RSI进行二次确认。仓位与风控决定投资比例、设置止损止盈位。一个完整的策略必须包含明确的出场规则例如当BSPR连续两日小于0.8时平仓或股价从买入点回撤超过8%时止损。交易成本模型中需考虑佣金和印花税对于高频或短线策略交易成本的影响尤为显著。5.4 性能优化与扩展思路如果股票池很大如全市场近5000只股票循环获取数据并计算会成为性能瓶颈。优化方法包括向量化计算尽可能使用pandas和numpy的向量化操作避免在Python层级的循环。并行处理使用multiprocessing或concurrent.futures库进行多进程/多线程计算充分利用多核CPU。数据本地化将常用的历史数据提前下载到本地数据库如SQLite、DuckDB或文件中避免每次运行都从网络接口重复获取。指标扩展买卖压力可以衍生出更多指标例如大单净流入只统计单笔成交金额超过一定阈值如50万元的主动性买卖差额。压力趋势计算BSPR的N日移动平均线观察其金叉死叉。压力背离股价创新低但BSPR指标未创新低底背离可能预示下跌动能衰竭。6. 一个完整的策略回测框架搭建思路为了科学地评估买卖压力策略的有效性我们需要进行历史回测。这里给出一个简易回测框架的搭建思路你可以基于此进行扩展。回测的核心是避免“未来函数”即只能用当时已经发生的数据做决策。我们的买卖压力数据是基于当日成交的在实盘中只能在收盘后才能完整计算。因此回测时我们假设在交易日T收盘后根据T日及之前的数据计算出信号在下一个交易日T1的开盘价执行买入/卖出操作。import pandas as pd import numpy as np class SimplePressureBacktester: def __init__(self, initial_capital1000000, commission_rate0.0003): self.initial_capital initial_capital self.commission_rate commission_rate self.positions {} # 持仓字典 {‘code’: shares} self.cash initial_capital self.equity_curve [] # 每日资产总值记录 def run_backtest(self, signal_df, price_df): 运行回测 :param signal_df: 信号DataFrame索引为日期列為股票代码值为1买入/持有或0卖出/不持有 :param price_df: 价格DataFrame索引为日期列為股票代码值为开盘价用于交易 dates signal_df.index.sort_values() prev_signal pd.Series(0, indexsignal_df.columns) # 前一日持仓信号 for date in dates: today_signal signal_df.loc[date] today_price price_df.loc[date] # 1. 处理卖出昨日有持仓但今日信号为0的股票 to_sell prev_signal[prev_signal 1].index.difference(today_signal[today_signal 1].index) for code in to_sell: if code in self.positions and code in today_price: sell_price today_price[code] sell_shares self.positions.pop(code) sell_amount sell_price * sell_shares self.cash sell_amount * (1 - self.commission_rate) # 扣除佣金 # 2. 处理买入今日信号为1但昨日未持仓的股票 # 简易策略等权重分配可用资金给所有新信号股票 to_buy today_signal[today_signal 1].index.difference(prev_signal[prev_signal 1].index) if not to_buy.empty: # 计算每只股票可分配资金扣除预计佣金 available_cash_per_stock self.cash / len(to_buy) for code in to_buy: if code in today_price: buy_price today_price[code] # 计算可买股数以100股为单位 max_shares int((available_cash_per_stock / (1 self.commission_rate)) // (buy_price * 100)) * 100 if max_shares 0: cost buy_price * max_shares * (1 self.commission_rate) self.positions[code] max_shares self.cash - cost # 3. 计算当日总资产 position_value 0 for code, shares in self.positions.items(): if code in today_price: position_value today_price[code] * shares total_equity self.cash position_value self.equity_curve.append({date: date, equity: total_equity}) # 4. 更新前一日信号 prev_signal today_signal.copy() # 回测结束平仓所有头寸假设最后一天以收盘价卖出 last_date dates[-1] last_price price_df.loc[last_date] for code, shares in list(self.positions.items()): if code in last_price: sell_amount last_price[code] * shares self.cash sell_amount * (1 - self.commission_rate) self.positions.clear() final_equity self.cash return pd.DataFrame(self.equity_curve).set_index(date), final_equity # 使用示例 # 1. 需要事先准备好 signal_df 和 price_df # signal_df: 每个交易日每只股票是否有买入/持有信号1/0 # price_df: 每个交易日每只股票的开盘价用于交易 # 2. 初始化回测器并运行 # backtester SimplePressureBacktester(initial_capital1000000) # equity_curve, final_value backtester.run_backtest(signal_df, price_df) # 3. 分析回测结果计算年化收益率、最大回撤、夏普比率等这个框架非常基础忽略了涨跌停无法买卖、滑点等问题但清晰地展示了回测的逻辑流程。在实际应用中你需要根据策略信号生成signal_df并获取对应的历史价格数据price_df。7. 我的实操心得与避坑指南在实践买卖压力策略的过程中我踩过不少坑也积累了一些心得这些是代码和公式里不会写的细节。心得一数据源的稳定性和一致性是生命线。早期我用过一些免费的网页爬虫接口经常因为网站改版而失效导致策略中断。后来我固定使用akshare这类维护相对活跃的库并养成了在策略核心代码外包裹严密异常捕获的习惯。对于关键数据我会定期抽样检查比如对比不同日期同一只股票的数据总量是否在合理范围内防止接口静默返回错误数据。永远不要完全信任你的数据源必须设置数据质量检查点。心得二买卖压力指标不是“圣杯”它揭示的是微观结构下的短期力量对比。我曾一度迷信高BSPR股票但发现很多情况下股价依然不涨。后来我意识到强大的买压可能发生在股价高位是主力对倒出货制造繁荣假象也可能发生在下跌初期是抄底资金与抛压的博弈但最终空头力量更强。因此必须将买卖压力指标放在更大的市场语境中结合大盘趋势、行业板块热度、个股所处的价格位置是突破关键位还是超跌反弹来综合判断。单独使用任何一个指标都是危险的。心得三参数过拟合是量化新手的通病。当我第一次回测时我不断调整BSPR的阈值1.1, 1.2, 1.3...、观察天数3天5天10天...直到在历史数据上获得一条漂亮的净值曲线。但这毫无意义因为这只是在拟合历史噪声。正确的做法是将数据分为样本内和样本外。用样本内数据例如2018-2020年确定一组“逻辑上合理”的参数比如BSPR1.2代表买压显著然后将其应用到完全未参与参数优化的样本外数据例如2021-2022年进行验证。如果策略在样本外依然有效才说明它可能具备一定的稳健性。心得四从想法到实盘风控的权重必须占到80%以上。选股模型只解决了“买什么”的问题而“买多少”、“何时卖”、“错了怎么办”才是决定长期盈亏的关键。在我的实盘框架中买卖压力信号只是一个触发因子它会进入一个包含仓位管理模块、风险平价模块和止损止盈模块的决策系统。例如即使某只股票BSPR信号很强如果当前整体市场风险偏高如波动率指数VIX飙升我的仓位管理模块也会强制降低单只股票的持仓上限。没有风控的量化策略就像没有刹车的跑车。最后我想强调的是本文提供的代码和思路是一个完整的、可运行的起点但它绝非一个“稳赚不赔”的赚钱机器。它的价值在于展示了如何将一种市场直觉感知买卖力量转化为可量化、可检验、可迭代的计算机程序。你可以在此基础上融入你自己的市场理解增加更多的过滤条件结合其他因子如估值、动量、质量等构建属于你自己的复合策略。量化交易的世界里没有一劳永逸的代码只有不断进化的认知和系统。
![[GESP202606 四级] 扫雷](http://pic.xiahunao.cn/yaotu/[GESP202606 四级] 扫雷)
