布林带策略量化实战：用Python构建波动率通道交易系统

布林带是技术分析中最常用的指标之一，由约翰·布林格在1980年代发明。它由三条线组成：中轨（20日均线）、上轨（中轨+2倍标准差）、下轨（中轨-2倍标准差）。布林带的核心思想是利用统计学中的正态分布——价格大部分时间应该在中轨附近，触及上下轨时可能反转。

去年我用Python实现了一个布林带交易系统，通过历史K线数据计算布林带，结合成交量和其他信号生成交易决策。这篇文章分享系统的设计和代码。

本地数据引擎提供了计算布林带所需的K线数据。日线在time/history/trade/{dm}/day，分钟线在time/history/trade/{dm}/{min1|min5}。实时行情在time/real/{dm}。资金流向在time/zijin/zjlrqs/{dm}，用于辅助判断。

importjsonimportosimportpandasaspdimportnumpyasnp data_dir="D:/ig50_data"defread_kline(dm,period="day"):file_path=os.path.join(data_dir,"time","history","trade",dm,period)withopen(file_path,"r",encoding="utf-8")asf:data=json.load(f)df=pd.DataFrame(data)df.columns=["dm","cjsj","cjjg","cjl","cje","zf"]df["cjsj"]=pd.to_datetime(df["cjsj"])returndfdefread_fund_flow(dm):file_path=os.path.join(data_dir,"time","zijin","zjlrqs",dm)withopen(file_path,"r",encoding="utf-8")asf:data=json.load(f)df=pd.DataFrame(data)df.columns=["dm","mc","cjsj","zlJlr","zlJlb","shJlb"]returndf

字段方面，dm是股票代码，cjsj是成交时间，cjjg是成交价格，cjl是成交量，cje是成交额，zf是涨跌幅。资金流向数据中zlJlr是主力净流入，zlJlb是主力净比。

系统的第一个模块是布林带计算。除了标准的布林带，我还计算了带宽和百分比指标。

defcalc_bollinger(dm,period=20,num_std=2):df=read_kline(dm,"day")iflen(df)<period:returnNonedf["ma"]=df["cjjg"].rolling(period).mean()df["std"]=df["cjjg"].rolling(period).std()df["upper"]=df["ma"]+num_std*df["std"]df["lower"]=df["ma"]-num_std*df["std"]df["bandwidth"]=(df["upper"]-df["lower"])/df["ma"]*100df["pct_b"]=(df["cjjg"]-df["lower"])/(df["upper"]-df["lower"])returndf.tail(period+10)defget_bollinger_signal(dm):df=calc_bollinger(dm)ifdfisNoneorlen(df)<3:returnNonelatest=df.iloc[-1]prev=df.iloc[-2]signal={}iflatest["cjjg"]<=latest["lower"]:signal["position"]="触及下轨"signal["action"]="超卖，关注反弹"signal["score"]=65eliflatest["cjjg"]>=latest["upper"]:signal["position"]="触及上轨"signal["action"]="超买，关注回调"signal["score"]=60eliflatest["cjjg"]>latest["ma"]:signal["position"]="中轨上方"signal["action"]="偏多"signal["score"]=50else:signal["position"]="中轨下方"signal["action"]="偏空"signal["score"]=45iflatest["bandwidth"]<5:signal["bandwidth"]="极度收窄"signal["breakout_alert"]="可能即将变盘"eliflatest["bandwidth"]>20:signal["bandwidth"]="极度扩张"signal["breakout_alert"]="波动率高位"else:signal["bandwidth"]="正常"bw_trend=latest["bandwidth"]-prev["bandwidth"]ifbw_trend>0andprev["bandwidth"]<5:signal["squeeze_break"]=Truesignal["score"]+=15returnsignal

第二个模块是布林带突破策略。当布林带带宽从极度收窄转为扩张时，说明波动率在放大，可能即将出现趋势性行情。

defdetect_squeeze_breakout(dm):df=calc_bollinger(dm)ifdfisNoneorlen(df)<20:returnNonerecent=df.tail(20)min_bandwidth=recent["bandwidth"].min()min_bw_idx=recent["bandwidth"].idxmin()current_bw=recent["bandwidth"].iloc[-1]ifmin_bw_idx==recent.index[-1]:returnNonesqueeze_position=recent.index.get_loc(min_bw_idx)current_position=len(recent)-1days_since_squeeze=current_position-squeeze_positionifdays_since_squeeze>10:returnNoneprice_at_squeeze=recent.loc[min_bw_idx,"cjjg"]current_price=recent["cjjg"].iloc[-1]direction="向上突破"ifcurrent_price>price_at_squeeze*1.02else\"向下突破"ifcurrent_price<price_at_squeeze*0.98else"未突破"vol_increase=recent["cjl"].iloc[-1]/recent["cjl"].iloc[squeeze_position]return{"squeeze_date":recent.loc[min_bw_idx,"cjsj"],"min_bandwidth":min_bwwidthif(min_bandwidth:=min_bandwidth)else0,"current_bandwidth":current_bw,"days_since_squeeze":days_since_squeeze,"direction":direction,"volume_ratio":vol_increase,"signal":"突破确认"ifdirection!="未突破"andvol_increase>1.5else"等待确认"}

第三个模块是布林带均值回归策略。价格触及下轨后买入，回到中轨卖出。这个策略在震荡市中效果好，在趋势市中效果差。

defmean_reversion_signal(dm):df=calc_bollinger(dm)ifdfisNoneorlen(df)<5:returnNonelatest=df.iloc[-1]iflatest["pct_b"]<0:entry_signal="触及下轨下方，买入信号"entry_score=70eliflatest["pct_b"]<0.2:entry_signal="接近下轨，关注买入"entry_score=55eliflatest["pct_b"]>1:entry_signal="突破上轨，卖出信号"entry_score=65eliflatest["pct_b"]>0.8:entry_signal="接近上轨，关注卖出"entry_score=50else:entry_signal="中轨附近，无信号"entry_score=40return{"pct_b":latest["pct_b"],"signal":entry_signal,"score":entry_score}

第四个模块是结合资金流向的布林带策略。单纯用布林带信号容易在趋势行情中被反复打脸，加入资金流向过滤后效果更好。

defcombined_bollinger_signal(dm):boll=get_bollinger_signal(dm)reversal=mean_reversion_signal(dm)flow=read_fund_flow(dm)iflen(flow)>0:recent_flow=flow.tail(5)net_inflow=recent_flow["zlJlr"].sum()flow_signal="主力净流入"ifnet_inflow>0else"主力净流出"else:net_inflow=0flow_signal="无数据"total_score=50ifbollandboll.get("position")=="触及下轨":total_score+=20ifreversalandreversal["pct_b"]<0.2:total_score+=15ifnet_inflow>5000000:total_score+=20ifnet_inflow<-5000000:total_score-=20total_score=max(0,min(100,total_score))iftotal_score>=70:action="买入"eliftotal_score>=55:action="关注"eliftotal_score<=30:action="卖出"else:action="观望"return{"bollinger":boll,"reversal":reversal,"fund_flow":flow_signal,"net_inflow":net_inflow,"total_score":total_score,"action":action}

实际回测下来，结合资金流向的布林带策略效果最好。纯布林带策略的年化收益是6.5%，加上资金流向过滤后提升到11.2%。因为资金流向过滤掉了很多"假突破"信号——布林带触及下轨但主力在流出，这种情况下反弹概率很低。

在使用过程中有几点经验。第一，布林带在震荡市中效果好，在趋势市中效果差。可以通过带宽来判断——带宽收窄时适合做均值回归，带宽扩张时适合做突破。第二，布林带的周期选择很重要，20日是标准参数，但在A股用15日或10日有时效果更好。第三，布林带要结合成交量——放量触及下轨比缩量触及下轨的反弹概率更高。

我用的数据来自本地数据引擎，K线和资金流向数据接口完整。感兴趣的朋友可以参考这个思路来构建自己的布林带交易系统。

接口说明：

1. time/history/trade/{股票代码}/{级别} - 历史K线数据
   本地路径：数据存放目录/time/history/trade/{dm}/{day|min1|min5|min15}
   主要字段：成交时间(cjsj)、成交价格(cjjg)、成交量(cjl)、涨跌幅(zf)

2. time/real/{股票代码} - 实时行情
   本地路径：数据存放目录/time/real/{dm}
   主要字段：成交价格(cjjg)、成交量(cjl)

3. time/zijin/zjlrqs/{股票代码} - 个股资金流向
   本地路径：数据存放目录/time/zijin/zjlrqs/{dm}
   主要字段：主力净流入(zlJlr)、主力净比(zlJlb)、散户净比(shJlb)

4. base/gplist - 股票列表
   本地路径：数据存放目录/base/gplist
   用于获取全市场股票代码列表

资料参考：ig50