- スクエア
- 海
海
作者: リン・ハーン
アウウウ, 2020年3月22日 23:19:46 日曜日
タグ:
import json
import time
class Turtle:
def __init__(self, account=None, donchian_channel_open_position=20, donchian_channel_stop_profit=10, atr_day_length=20, max_risk_ratio=0.5):
self.donchian_channel_open_position = donchian_channel_open_position # 唐奇安通道的天数周期(开仓)
self.donchian_channel_stop_profit = donchian_channel_stop_profit # 唐奇安通道的天数周期(止盈)
self.atr_day_length = atr_day_length # ATR计算所用天数
self.max_risk_ratio = max_risk_ratio # 最高风险度
self.state = {
"position": 0, # 本策略净持仓数(正数表示多头，负数表示空头，0表示空仓)
"last_price": float("nan"), # 上次调仓价
}
positions = _C(exchange.GetPosition)
self.equity=0
for position in positions:
if position["Type"]==PD_LONG:
self.state["position"]=position["Amount"]
self.state["last_price"] = position["Price"]
self.equity+=position["Margin"]
elif position["Type"]==PD_SHORT:
self.state["position"]=-position["Amount"]
self.state["last_price"] = position["Price"]
self.equity+=position["Margin"]
self.account = _C(exchange.GetAccount)
self.equity += self.account["Stocks"]+self.account["FrozenStocks"]
#Log(self.equity)
self.n = 0 # 平均真实波幅(N值)
self.unit = 0 # 买卖单位
self.donchian_channel_high = 0 # 唐奇安通道上轨
self.donchian_channel_low = 0 # 唐奇安通道下轨
# 由于ATR是路径依赖函数，因此使用更长的数据序列进行计算以便使其值稳定下来
self.klines = exchange.GetRecords()
def recalc_paramter(self):
# 平均真实波幅(N值)
self.equity=0
positions = _C(exchange.GetPosition)
for position in positions:
if position["Type"]==PD_LONG:
self.equity+=position["Margin"]
elif position["Type"]==PD_SHORT:
self.equity+=position["Margin"]
self.account = _C(exchange.GetAccount)
self.equity += self.account["Stocks"]+self.account["FrozenStocks"]
#Log(self.equity)
records = _C(exchange.GetRecords)
self.n =TA.ATR(records, self.atr_day_length)[-1]
# 买卖单位
self.current_price = records[-1]["Close"]
self.unit = int((self.equity * 0.01*self.current_price*self.current_price) / (100 * self.n))
# 唐奇安通道上轨：前N个交易日的最高价
#Log(records)
self.donchian_channel_high =TA.Highest(records, self.donchian_channel_open_position , 'High') #唐奇安通道上轨：前N个交易日的最高价
self.donchian_channel_high =TA.Highest(records, 55 , 'High')
# 唐奇安通道下轨：前N个交易日的最低价
self.donchian_channel_low = TA.Lowest(records, self.donchian_channel_open_position , 'Low')
self.donchian_channel_low = TA.Lowest(records, 55 , 'Low')
#Log("唐其安通道上下轨: %f, %f" % (self.donchian_channel_high, self.donchian_channel_low))
self.stop_high = TA.Highest(records, self.donchian_channel_stop_profit , 'High')
self.stop_high = TA.Highest(records, 20 , 'High')
self.stop_low = TA.Highest(records, self.donchian_channel_stop_profit , 'Low')
self.stop_low = TA.Highest(records, 20, 'Low')
boll = TA.BOLL(records, 50, 2)
self.up_line = boll[0][-1]
self.mid_line = boll[1][-1]
self.down_line = boll[2][-1]
close1 = records[-2]['Close'] # 最新收盘价
close30 = records[-30]['Close'] # 前30根K线的收盘价
hh30 = TA.Highest(records, 30, 'High') # 最近30根K线的最高价
ll30 = TA.Lowest(records, 30, 'Low') # 最近30根K线的最低价
self.cmi = abs((close1 - close30) / (hh30 - ll30)) * 100 # 计算市场波动指数
return True
def set_position(self, pos):
self.state["position"] = pos
self.state["last_price"] = self.current_price
positions = _C(exchange.GetPosition)
sell_amount =0
long_amount = 0
for position in positions:
if position["Type"]==PD_LONG:
long_amount=position["Amount"]
elif position["Type"]==PD_SHORT:
sell_amount=position["Amount"]
if pos>0:
if sell_amount>0:
exchange.SetDirection("closesell")
exchange.Buy(self.current_price*1.005,sell_amount)
if pos>long_amount:
exchange.SetDirection("buy")
exchange.Buy(self.current_price*1.005,pos-long_amount)
elif pos<long_amount:
exchange.SetDirection("closebuy")
exchange.Sell(self.current_price*0.995,long_amount-pos)
elif pos<0:
pos=-pos
if long_amount>0:
exchange.SetDirection("closebuy")
exchange.Sell(self.current_price*0.995,long_amount)
if pos>sell_amount:
exchange.SetDirection("sell")
exchange.Sell(self.current_price*0.995,pos-sell_amount)
elif pos<sell_amount:
exchange.SetDirection("closesell")
exchange.Buy(self.current_price*1.005,sell_amount-pos)
else:
if long_amount>0:
exchange.SetDirection("closebuy")
exchange.Sell(self.current_price*0.995,long_amount)
if sell_amount>0:
exchange.SetDirection("closesell")
exchange.Buy(self.current_price*1.005,sell_amount)
#self.target_pos.set_target_volume(self.state["position"])
def try_open(self):
"""开仓策略"""
while self.state["position"] == 0:
self.recalc_paramter()
#Log("最新价: %f" % self.current_price)
if self.current_price > self.donchian_channel_high: # 当前价>唐奇安通道上轨，买入1个Unit；(持多仓)
#if self.cmi>20 and self.current_price>self.up_line:
#Log("当前价>唐奇安通道上轨，买入1个Unit(持多仓): %d 手" % self.unit)
self.set_position(self.state["position"] + self.unit)
elif self.current_price < self.donchian_channel_low: # 当前价<唐奇安通道下轨，卖出1个Unit；(持空仓)
#elif self.cmi>20 and self.current_price<self.down_line:
#Log("当前价<唐奇安通道下轨，卖出1个Unit(持空仓): %d 手" % self.unit)
self.set_position(self.state["position"] - self.unit)
def try_close(self):
"""交易策略"""
while self.state["position"] != 0:
if True:
self.recalc_paramter()
Log("最新价: ", self.current_price)
#if self.cmi<20:
# self.set_position(0)
if self.state["position"] > 0: # 持多单
# 加仓策略: 如果是多仓且行情最新价在上一次建仓（或者加仓）的基础上又上涨了0.5N，就再加一个Unit的多仓,并且风险度在设定范围内(以防爆仓)
if self.current_price >= self.state["last_price"] + 0.5 * self.n and self.state["position"] + self.unit<=4*self.unit:
Log("加仓:加1个Unit的多仓")
self.set_position(self.state["position"] + self.unit)
# 止损策略: 如果是多仓且行情最新价在上一次建仓（或者加仓）的基础上又下跌了2N，就卖出全部头寸止损
elif self.current_price <= self.state["last_price"] - 2 * self.n:
Log("止损:卖出全部头寸")
self.set_position(0)
# 止盈策略: 如果是多仓且行情最新价跌破了10日唐奇安通道的下轨，就清空所有头寸结束策略,离场
if self.current_price <= self.stop_low:
#if self.current_price<self.mid_line:
Log("止盈:清空所有头寸结束策略,离场")
self.set_position(0)
elif self.state["position"] < 0: # 持空单
# 加仓策略: 如果是空仓且行情最新价在上一次建仓（或者加仓）的基础上又下跌了0.5N，就再加一个Unit的空仓,并且风险度在设定范围内(以防爆仓)
if self.current_price <= self.state["last_price"] - 0.5 * self.n and (-self.state["position"]) + self.unit<=4*self.unit:
Log("加仓:加1个Unit的空仓")
self.set_position(self.state["position"] - self.unit)
# 止损策略: 如果是空仓且行情最新价在上一次建仓（或者加仓）的基础上又上涨了2N，就平仓止损
elif self.current_price >= self.state["last_price"] + 2 * self.n:
Log("止损:卖出全部头寸")
self.set_position(0)
# 止盈策略: 如果是空仓且行情最新价升破了10日唐奇安通道的上轨，就清空所有头寸结束策略,离场
if self.current_price >= self.stop_high:
#if self.current_price>self.mid_line:
Log("止盈:清空所有头寸结束策略,离场")
self.set_position(0)
def strategy(self):
"""海龟策略"""
Log("等待K线及账户数据...")
while not self.recalc_paramter():
raise Exception("获取数据失败，请确认行情连接正常并已经登录交易账户")
while True:
self.try_open()
self.try_close()
def main():
exchange.SetContractType("quarter")
turtle = Turtle(donchian_channel_open_position=Donchian_open,donchian_channel_stop_profit=Donchian_stop,atr_day_length=Atr)
Log("策略开始运行")
Log("当前持仓数: %d, 上次调仓价: %f" % (turtle.state["position"], turtle.state["last_price"]))
turtle.strategy()
もっと