使用 tqsdk 进行期货交易
输入“/”快速插入内容
使用 tqsdk 进行期货交易
使用 tqsdk 进行期货交易
飞书用户7395
飞书用户56332023年4月14日创建
3898
5354
6
0
•
代码块
import pandas as pd
from loguru import logger
from datetime import date, datetime, timedelta
from tqsdk import TqApi, TqAuth, TqSim, TqBacktest, TargetPosTask, BacktestFinished
from czsc import Freq, RawBar
from czsc.strategies import CzscStrategyExample2 as Strategy
def format_kline(df, freq=Freq.F1):
"""对分钟K线进行格式化"""
freq = Freq(freq)
rows = df.to_dict('records')
raw_bars = []
for i, row in enumerate(rows):
bar = RawBar(symbol=row['symbol'], id=i, freq=freq,
dt=datetime.fromtimestamp(row["datetime"] / 1e9) + timedelta(minutes=1),
open=row['open'], close=row['close'], high=row['high'],
low=row['low'], vol=row['volume'], amount=row['volume'] * row['close'])
raw_bars.append(bar)
return raw_bars
def run_multi_main_symbol(**kwargs):
"""执行多个主力连续合约的策略"""
tq_user = kwargs.get("tq_user")
tq_pwd = kwargs.get("tq_pwd")
init_money = int(kwargs.get("init_money", 1000000))
sdt = kwargs.get("sdt", "20180101")
edt = kwargs.get("edt", "20230414")
sdt = pd.to_datetime(sdt).date()
edt = pd.to_datetime(edt).date()
api = TqApi(TqSim(init_money), web_gui=True, auth=TqAuth(tq_user, tq_pwd),
backtest=TqBacktest(start_dt=sdt, end_dt=edt))
symbols = ["KQ.m@DCE.i", "KQ.m@DCE.j", "KQ.m@DCE.jm", "KQ.m@SHFE.hc", "KQ.m@SHFE.rb"]
# 货值等权配置手数
pos_multiplier = {"KQ.m@DCE.i": 3, "KQ.m@DCE.j": 1, "KQ.m@DCE.jm": 2, "KQ.m@SHFE.hc": 5, "KQ.m@SHFE.rb": 5}
metas = {symbol: {} for symbol in symbols}
for symbol in symbols:
tactic = Strategy(symbol=symbol)
kline = api.get_kline_serial(symbol, int(tactic.base_freq.strip('分钟')) * 60, data_length=10000, adj_type='B')
quote = api.get_quote(symbol)
raw_bars = format_kline(kline, freq=tactic.base_freq)
trader = tactic.init_trader(raw_bars, sdt=date(2017, 10, 1))
target_pos = TargetPosTask(api, quote.underlying_symbol)
metas[symbol] = {
"symbols": symbol,
"kline": kline,
"quote": quote,
"trader": trader,
"base_freq": tactic.base_freq,
"target_pos": target_pos
}
try:
while api.wait_update():
for symbol, meta in metas.items():
kline = meta["kline"]
quote = meta["quote"]
trader = meta["trader"]
target_pos = meta["target_pos"]
if api.is_changing(quote, "underlying_symbol"):
# 主力连续合约换月
logger.info(f"主力换月：{quote.datetime} - {quote.underlying_symbol}")
target_pos.set_target_volume(0)
target_pos = TargetPosTask(api, quote.underlying_symbol)
if api.is_changing(kline.iloc[-1], "datetime"):
new_bars = format_kline(kline.tail(10), freq=meta['base_freq'])
new_bars = [x for x in new_bars if x.dt > trader.end_dt]
for bar in new_bars:
trader.update(bar)
target_pos.set_target_volume(trader.get_ensemble_pos('vote') * pos_multiplier[symbol])
metas[symbol] = {
"symbols": symbol,
"kline": kline,
"quote": quote,
"trader": trader,
"base_freq": meta['base_freq'],
"target_pos": target_pos
}
except BacktestFinished as e:
while True:
api.wait_update()
if __name__ == '__main__':
run_multi_main_symbol()