显示报错：
from tqsdk import TqApi, TqAuth, TqBacktest
from tqsdk.tafunc import ma
from tqsdk.exceptions import TqTimeoutError
import datetime as dt
import numpy as np
# 策略配置参数
# 策略配置参数
STRATEGY_CONFIG = {
"underlying": "CFFEX.IM2403", # 标的期货合约（中证1000期货）
"option_underlying": "MO", # 期权标的代码
"backtest_start": dt.datetime(2024, 1, 1),
"backtest_end": dt.datetime(2024, 3, 31),
"vol_window": 60,
"strike_offset": 100,
"lots": 1,
"profit_target": 1500,
"loss_limit": -3000,
"open_time_start": dt.time(9, 30),
"open_time_end": dt.time(11, 0),
"close_time": dt.time(14, 45)
}
class OptionVolatilityStrategy:
def __init__(self, api, underlying_symbol):
self.api = api
self.underlying_symbol = underlying_symbol
self.underlying_quote = None # 延迟初始化
self.atm_option_id = None
self.vol_klines = None
self.trading_state = {
"position": False,
"entry_price": {"call": 0, "put": 0},
"options": {"call": None, "put": None},
"total_capital": 0,
"last_trading_day": ""
}
self._init_strategy()
def _init_strategy(self):
"""初始化策略（移除get_trading_days依赖）"""
self.trading_state["total_capital"] = self._get_account_balance()
self.atm_option_id = self._find_atm_option()
self.vol_klines = self._init_vol_klines()
# 简化：不依赖交易日历，直接使用回测时间范围
self.trading_days = []
print(f"策略初始化完成，回测时段: {STRATEGY_CONFIG['backtest_start']} 到 {STRATEGY_CONFIG['backtest_end']}")
def _get_account_balance(self):
account = self.api.get_account()
return account.available if account else 100000
def _find_atm_option(self):
"""查找平值期权"""
try:
# 使用期货合约代码查询对应的期权合约
print(f"使用期货合约 {STRATEGY_CONFIG['underlying']} 查询期权合约...")
all_options = self.api.query_quotes(STRATEGY_CONFIG["underlying"])
if not all_options:
print(f"错误：未找到{STRATEGY_CONFIG['underlying']}对应的期权合约")
return None
print(f"成功找到期权合约: 共{len(all_options)}个")
# 获取期货合约价格作为标的价格
underlying_quote = self.api.get_quote(STRATEGY_CONFIG["underlying"])
underlying_price = underlying_quote.last_price
print(f"期货合约价格: {underlying_price}")
# 筛选平值看涨期权（行权价最接近当前价格的看涨期权）
call_options = [opt for opt in all_options if opt.option_class == "CALL"]
if not call_options:
print("错误：未找到看涨期权")
return None
# 找到行权价最接近当前价格的看涨期权
atm_option = min(call_options, key=lambda x: abs(x.strike_price - underlying_price))
print(f"找到平值看涨期权: {atm_option.instrument_id}, 行权价: {atm_option.strike_price}")
return atm_option.instrument_id
except Exception as e:
print(f"平值期权查找失败: {str(e)}")
return None
def _init_vol_klines(self):
if not self.atm_option_id:
return None
return {
"daily": self.api.get_kline_serial(self.atm_option_id, 86400, data_length=80),
"15m": self.api.get_kline_serial(self.atm_option_id, 900, data_length=30)
}
def _check_contract_expiry(self):
if not self.atm_option_id:
return False
try:
option_info = self.api.get_instrument(self.atm_option_id)
expiry_date = dt.datetime.strptime(option_info.expire_date, "%Y%m%d")
current_trading_day = self.api.get_trading_day()
return (expiry_date - current_trading_day).days <= 5
except Exception as e:
print(f"合约到期检查失败: {str(e)}")
return True
def _calculate_volatility_ma(self):
if not self.vol_klines:
return None, None
if len(self.vol_klines["daily"]) < STRATEGY_CONFIG["vol_window"]:
print(f"日线数据不足: {len(self.vol_klines['daily'])}根")
return None, None
vol_ma = ma(self.vol_klines["daily"]["close"], STRATEGY_CONFIG["vol_window"]).iloc[-1]
current_vol = self.vol_klines["daily"]["close"].iloc[-1]
return current_vol, vol_ma
def _select_out_of_money_options(self):
"""选择虚值期权"""
try:
# 使用期货合约代码查询对应的期权合约
print(f"使用期货合约 {STRATEGY_CONFIG['underlying']} 查询期权合约...")
all_options = self.api.query_options(STRATEGY_CONFIG["underlying"])
if not all_options:
print(f"错误：未找到{STRATEGY_CONFIG['underlying']}对应的期权合约")
return None, None
print(f"成功找到期权合约: 共{len(all_options)}个")
# 获取期货合约价格作为标的价格
underlying_quote = self.api.get_quote(STRATEGY_CONFIG["underlying"])
underlying_price = underlying_quote.last_price
print(f"期货合约价格: {underlying_price}")
target_call = underlying_price + STRATEGY_CONFIG["strike_offset"]
target_put = underlying_price - STRATEGY_CONFIG["strike_offset"]
print(f"目标看涨期权行权价: {target_call}, 目标看跌期权行权价: {target_put}")
# 筛选看涨期权和看跌期权
call_options = [opt for opt in all_options if opt.option_class == "CALL"]
put_options = [opt for opt in all_options if opt.option_class == "PUT"]
if not call_options or not put_options:
print("错误：未找到看涨或看跌期权")
return None, None
# 找到最接近目标行权价的看涨期权（虚值）
call_candidate = min(call_options, key=lambda x: abs(x.strike_price - target_call))
# 找到最接近目标行权价的看跌期权（虚值）
put_candidate = min(put_options, key=lambda x: abs(x.strike_price - target_put))
print(f"选择虚值看涨期权: {call_candidate.instrument_id}, 行权价: {call_candidate.strike_price}")
print(f"选择虚值看跌期权: {put_candidate.instrument_id}, 行权价: {put_candidate.strike_price}")
return call_candidate.instrument_id, put_candidate.instrument_id
except Exception as e:
print(f"选择虚值期权失败: {str(e)}")
return None, None
def _validate_option_contract(self, options):
if not options:
return None
current_trading_day = self.api.get_trading_day()
for opt in options:
expiry_date = dt.datetime.strptime(opt.expire_date, "%Y%m%d")
if expiry_date > current_trading_day:
return opt
print("未找到有效期权合约")
return None
def _check_time_window(self, current_time):
return STRATEGY_CONFIG["open_time_start"] <= current_time <= STRATEGY_CONFIG["open_time_end"]
def _check_daily_init_condition(self):
current_trading_day = self.api.get_trading_day().strftime("%Y%m%d")
if current_trading_day != self.trading_state["last_trading_day"]:
self.trading_state["last_trading_day"] = current_trading_day
return True
return False
def run(self):
print("策略开始回测...")
while True:
try:
if not self.api.wait_update():
break
current_bar_time = self.vol_klines["15m"].datetime.iloc[-1] if self.vol_klines else None
if not current_bar_time:
continue
current_time = dt.datetime.fromtimestamp(current_bar_time / 1e9).time()
current_date = dt.datetime.fromtimestamp(current_bar_time / 1e9).date()
if self._check_daily_init_condition():
print(f"\n=== 新交易日初始化: {self.trading_state['last_trading_day']} ===")
self.trading_state["total_capital"] = self._get_account_balance()
self.atm_option_id = self._find_atm_option()
self.vol_klines = self._init_vol_klines()
self.trading_state["position"] = False
if self._check_contract_expiry():
print("期权合约即将到期，更换合约...")
self.atm_option_id = self._find_atm_option()
self.vol_klines = self._init_vol_klines()
if not self.trading_state["position"]:
if self._check_time_window(current_time):
current_vol, vol_ma = self._calculate_volatility_ma()
if current_vol and vol_ma:
vol_condition = current_vol > vol_ma * 1.05
print(f"[{current_date} {current_time}] 波动率条件: 当前{current_vol:.4f}, 均线{vol_ma:.4f}, 满足={vol_condition}")
if len(self.vol_klines["15m"]) >= 2:
last_bar = self.vol_klines["15m"].iloc[-2]
kline_condition = last_bar.close < last_bar.open
print(f"[{current_date} {current_time}] K线条件: 开盘{last_bar.open:.2f}, 收盘{last_bar.close:.2f}, 收阴={kline_condition}")
if vol_condition and kline_condition:
call_opt, put_opt = self._select_out_of_money_options()
if call_opt and put_opt:
call_quote = self.api.get_quote(call_opt)
put_quote = self.api.get_quote(put_opt)
self.api.insert_order(
call_opt, "SELL", "OPEN",
STRATEGY_CONFIG["lots"],
call_quote.bid_price1
)
self.api.insert_order(
put_opt, "SELL", "OPEN",
STRATEGY_CONFIG["lots"],
put_quote.bid_price1
)
self.trading_state.update({
"position": True,
"options": {"call": call_opt, "put": put_opt},
"entry_price": {
"call": call_quote.bid_price1,
"put": put_quote.bid_price1
}
})
print(f"[{current_date} {current_time}] 提交开仓订单: CALL={call_opt}, PUT={put_opt}")
if self.trading_state["position"]:
if current_time >= STRATEGY_CONFIG["close_time"]:
self._close_position()
print(f"[{current_date} {current_time}] 到达收盘时间，强制平仓")
total_pnl = self._calculate_pnl()
if total_pnl >= STRATEGY_CONFIG["profit_target"] or total_pnl <= STRATEGY_CONFIG["loss_limit"]:
self._close_position()
print(f"[{current_date} {current_time}] 达到盈亏条件，平仓。当前盈亏: {total_pnl:.2f}")
except TqTimeoutError:
print("数据更新超时")
except Exception as e:
print(f"策略运行异常: {str(e)}")
if self.trading_state["position"]:
self._close_position()
break
print("回测结束")
def _calculate_pnl(self):
if not self.trading_state["options"]["call"] or not self.trading_state["options"]["put"]:
return 0
call_pos = self.api.get_position(self.trading_state["options"]["call"])
put_pos = self.api.get_position(self.trading_state["options"]["put"])
return call_pos.float_profit + put_pos.float_profit
def _close_position(self):
if self.trading_state["options"]["call"]:
call_quote = self.api.get_quote(self.trading_state["options"]["call"])
self.api.insert_order(
self.trading_state["options"]["call"], "BUY", "CLOSE",
STRATEGY_CONFIG["lots"],
call_quote.ask_price1
)
if self.trading_state["options"]["put"]:
put_quote = self.api.get_quote(self.trading_state["options"]["put"])
self.api.insert_order(
self.trading_state["options"]["put"], "BUY", "CLOSE",
STRATEGY_CONFIG["lots"],
put_quote.ask_price1
)
self.trading_state["position"] = False
if __name__ == "__main__":
backtest_start = STRATEGY_CONFIG["backtest_start"]
backtest_end = STRATEGY_CONFIG["backtest_end"]
from tqsdk import TqKq
api = TqApi(
TqKq(),
backtest=TqBacktest(start_dt=backtest_start, end_dt=backtest_end),
auth=TqAuth("xiaonangua1028", "xiaonangua1028"),
web_gui=True
)
try:
strategy = OptionVolatilityStrategy(api, STRATEGY_CONFIG["underlying"])
strategy.run()
finally:
api.close()
“D:\Productivity Tools\Python\python.exe” “D:\Productivity Tools\Pycharm\Codes\量化\tqsdk_option.py”
在使用天勤量化之前，默认您已经知晓并同意以下免责条款，如果不同意请立即停止使用：https://www.shinnytech.com/blog/disclaimer/
INFO – 您可以访问 http://127.0.0.1:51053 查看策略绘制出的 K 线图形。
INFO – 通知 xiaonangua1028: 登录成功
INFO – 通知 xiaonangua1028: 与 wss://otg-sim.shinnytech.com/trade 的网络连接已建立
使用期货合约 CFFEX.IM2403 查询期权合约…
平值期权查找失败: Class does not accept value CFFEX.IM2403
策略初始化完成，回测时段: 2024-01-01 00:00:00 到 2024-03-31 00:00:00
策略开始回测…
chaos 已回答的问题 2025年10月27日
STRATEGY_CONFIG = {
“underlying”: “CSI.000852”, # 标的股票指数 – 中证1000指数
“option_class”: “MO”, # 期权品类 – 中金所中证1000股指期权
“backtest_start”: dt.datetime(2024, 1, 1),
“backtest_end”: dt.datetime(2024, 3, 31),
“vol_window”: 60, # 波动率窗口
“strike_offset”: 100, # 行权价偏移点数
“lots”: 1, # 手数
“profit_target”: 1500, # 止盈
“loss_limit”: -3000, # 止损
“open_time_start”: dt.time(9, 30),
“open_time_end”: dt.time(11, 0),
“close_time”: dt.time(14, 45)
}
大佬，我这边进行了如上的修改，为啥还是不行，可以帮解答一次吗，问了很多AI模型都不行，感谢您