python股票均线策略模板

Willows · 发表于 2024-1-4 16:08:51

本示例仅作策略演示和模板介绍，请不要直接用于实际交易

策略导入的方式
1. 直接复制到空代码文件
2. 将.rzrk附件导入到QMT/投研版内部

策略讲解

双均线策略是一种基于技术分析的交易策略，主要用于研判股票价格走势。该策略以两条不同周期的移动平均线（例如5日线与30日线，即短期均线和长期均线）的交叉情况作为买入或卖出信号。

以下是其中一些重要的专业概念：
移动平均线 (Moving Average, MA): 这是衡量股票在特定时间范围内平均价格的一种指标。它可以帮助投资者确定市场的趋势。
金叉: 当短期均线从下方穿过长期均线时，被称为“金叉”。这通常被视为一个买入信号，因为它可能预示着一个上涨趋势的开始。
死叉: 当短期均线从上方穿过长期均线时，被称为“死叉”。这通常被视为一个卖出信号，因为它可能预示着一个下跌趋势的开始。

尽管双均线策略在一些市场环境中能提供相对准确的交易信号，但它并非无懈可击。例如，在震荡市中，频繁发生的假信号可能导致不必要的交易和潜在损失。因此，投资者通常会配合其他技术分析工具（如相对强弱指标RSI、布林带等）或基本面信息来优化策略与决策。

回测结果

策略源码

# coding:gbk
# 单策略独立进程是指每个策略拥有自己独立的进程，策略之间互不影响
# 在qmt内部使用单策略独立进程模式需要勾选上右侧的独立python进程
# 也在外部IDE使用该模式，此时需要把bin.x64/Lib/site-packages/xtquant复制到自己的python环境变量下
class _a():
pass
A = _a()
def init(C):
A.symbol = "510050.SH" # 填一个具体合约
A.period = C.period
A.line1 = 20 # 第一根线的参数
A.line2 = 40 # 第二根线的参数
C.accID = "123123" # 回测时需要使用这一句
# C.accID = account # 实盘时需要使用这一句
C.main_stock= C.stockcode + '.' +C.market
return
def after_init(C):
# 不确定有没有数据的时候，下载一遍数据
if 1:
my_download([A.symbol,C.main_stock],period = A.period)
def handlebar(C):
# print(C.barpos)
backTestTime = timetag_to_datetime(C.get_bar_timetag(C.barpos),"%Y%m%d%H%M%S")
print(backTestTime)
# 数据周期与右侧默认周期一致
kline = C.get_market_data_ex(['open', 'high','low','close'],[A.symbol],period = A.period, end_time = backTestTime,count = 3 + max([A.line1, A.line2]))[A.symbol]
ma1 = kline["close"].rolling(A.line1).mean()
ma2 = kline["close"].rolling(A.line2).mean()
holdings = get_stock_holdings(C.accID)
lots = holdings.get(A.symbol,{}).get("持仓数量",0)
avb_lots = holdings.get(A.symbol,{}).get("可用余额",0)
print(f"已走完K线 --- ma1:{ma1.iloc[-2]} ma2:{ma2.iloc[-2]}" )
print(f"最新K线 --- ma2:{ma1.iloc[-1]} ma2:{ma2.iloc[-1]}" )
cross_up = (ma1.iloc[-2] > ma2.iloc[-2]) and (ma1.iloc[-3] < ma2.iloc[-3])
cross_down = (ma1.iloc[-2] < ma2.iloc[-2]) and (ma1.iloc[-3] > ma2.iloc[-3])
# 如果有新的条件需要满足,可以在if语句用and进行连接
if cross_up and lots == 0 :#
print(f"ma1上穿ma2")
my_passorder(C,A.symbol, "buy",10000)
elif cross_down and avb_lots > 0:#
print(f"ma1下穿ma2")
my_passorder(C,A.symbol, "sell", lots)
else:
print(holdings)
return
# --------------------股票版本------------------------------
def my_passorder(C,stock,opentype,lots,price = None,m_strRemark = '系统备注'):
'''
Args:
C: ContextInfo
stock: 股票代码
opentype:
'buy' 买股票
'sell' 卖股票
lots: 股数
price: 下单价格，不指定时默认按对手价下单
m_strRemark = '系统备注' 用于自定义寻找orderID
'''
#holdings = get_holdings(C.accID,'stock')
opentype = opentype #买卖方向
op =1101 #股数
opType = 23 if opentype == 'buy' else 24# 若不为'sell' 则为buy
volumex = lots
price = 0 if not price else price # price参数必须存在
prType = 14 if not price else 11 # 若不指定价格，则默认按对手价下单
print(f'{stock} 新委托信息方向{opentype} 价格{price} 量{volumex}')
#print(f"opType:{opType} , op:{op} , C.accID{C.accID} , stock{stock} , prType{prType} , price{price} , volumex{volumex}")
passorder(opType, op, C.accID,stock, prType, price, volumex,'交易注释',1,'{}'.format(m_strRemark), C)
print(f'委托发送完成')
def get_stock_holdings(accid,symbol = None):
'''
Arg:
accondid:账户id
datatype:'FUTURE'：期货,'STOCK'：股票,......
symbol: 品种，不填默认返会全部持仓
return:
{股票代码:{'手数':int,"持仓成本":float,'浮动盈亏':float,"可用余额":int}}
'''
PositionInfo_dict={}
resultlist = get_trade_detail_data(accid,"STOCK",'POSITION')
for obj in resultlist:
PositionInfo_dict[obj.m_strInstrumentID+"."+obj.m_strExchangeID] = {
"持仓数量":obj.m_nVolume,
"持仓成本":obj.m_dOpenPrice,
"浮动盈亏":obj.m_dFloatProfit,
"盈亏比例":obj.m_dProfitRate * 100,
"市值":obj.m_dMarketValue,
"可用余额":obj.m_nCanUseVolume,
"交易方向":obj.m_nDirection
}
if symbol:
return PositionInfo_dict[symbol]
else :
return PositionInfo_dict
def my_download(stock_list,period,start_date = '', end_date = ''):
'''
用于显示下载进度
'''
if "d" in period:
period = "1d"
elif "m" in period:
if int(period[0]) < 5:
period = "1m"
else:
period = "5m"
elif "tick" == period:
pass
else:
raise KeyboardInterrupt("周期传入错误")
n = 1
num = len(stock_list)
for i in stock_list:
print(f"当前正在下载{n}/{num}")
download_history_data(i,period,start_date, end_date)
n += 1
print("下载任务结束")