回测能正常运行，但是只交易了几次就停了，止损止盈也未触发。
由bqbkm8ac创建，最终由bqbkm8ac 被浏览 5 用户
from bigmodule import M
# 交易引擎：初始化函数，只执行一次
def m5_initialize_bigquant_run(context):
from bigtrader.finance.commission import PerOrder
context.set_commission(PerOrder(buy_cost=0.0003, sell_cost=0.0013, min_cost=5))
context.first_run = True # 初次运行标志
context.stop_buy_days = 0 # 停止购买天数
context.hold_days = 0 # 持仓天数计数
# @param(id="m7", name="handle_data")
def m5_handle_data_bigquant_run(context, data):
import pandas as pd
# 获取当前时间的数据
current_data = context.data[context.data["date"] == data.current_dt.strftime("%Y-%m-%d")]
if context.first_run:
context.first_run = False
context.sorted_stocks = current_data.sort_values('volume_ratio', ascending=False)
target_instruments = set(context.sorted_stocks["instrument"][:5])
# 购买评分最高前5只股票
for stock in target_instruments:
context.order_target_value(stock, context.portfolio.cash * 0.2)
else:
current_positions = list(context.get_account_positions().keys())
sorted_positions = current_data[current_data["instrument"].isin(current_positions)].sort_values('volume_ratio', ascending=False)
stocks_to_sell = []
# 止盈、止损与动态回撤止损
for stock in current_positions:
position = context.get_position(stock)
current_price = data.current([stock], 'price')[stock]
return_pct = (current_price - position.cost_price) / position.cost_price if position.cost_price != 0 else 0
max_price = position.last_sale_price
drawdown = (current_price - max_price) / max_price if max_price != 0 else 0
# 动态止盈
if return_pct >= 0.2:
context.order_target_value(stock, position.amount * 0.5 * current_price)
context.stop_buy_days = 5
# 止损：跌幅超过 -10%时全部清仓
elif return_pct <= -0.1:
context.order_target_percent(stock, 0)
stocks_to_sell.append(stock)
# 动态回撤止损
elif drawdown <= -0.08:
context.order_target_percent(stock, 0)
stocks_to_sell.append(stock)
context.stop_buy_days = 5
# 卖出后买入评分最高的
if context.stop_buy_days == 0 and not sorted_positions.empty:
to_sell = sorted_positions["instrument"].iloc[-1] # 持仓中评分最低的股票
if to_sell not in stocks_to_sell:
context.order_target_percent(to_sell, 0)
remaining_cash =context.portfolio.cash
if remaining_cash > 0:
sorted_scores = current_data.sort_values('volume_ratio', ascending=False)
top_buy = sorted_scores["instrument"].iloc[0] #市场中评分最高的股票
if top_buy not in current_positions:
context.order_target_value(top_buy, remaining_cash)
# 每日减少停止买入天数
context.stop_buy_days = max(0, context.stop_buy_days - 1)
# 更新持仓天数
context.hold_days += 1
m1 = M.cn_stock_basic_selector.v8(
exchanges=["""上交所""", """深交所"""],
list_sectors=["""主板""", """创业板""", """科创板"""],
indexes=["""中证1000"""],
st_statuses=["""正常"""],
drop_suspended=True,
m_name="""m1"""
)
m2 = M.input_features_dai.v30(
expr="m_lag(volume, 1) / m_avg(m_lag(volume, 1), 5) AS volume_ratio,"\
"m_lag(volume, 2) / m_avg(m_lag(volume, 2), 5) AS volume_ratio_prev1,"\
"c_quantile_cont(close, 0.4) AS price_30th_percentile",
expr_filters="volume_ratio > 1.2 AND volume_ratio < 1.8 "\
"AND volume_ratio_prev1 > 1.3 AND volume_ratio_prev1 < 1.8 "\
"AND close < price_30th_percentile AND st_status = 0",
expr_tables="cn_stock_prefactors",
sql=m1.data,
expr_drop_na=True,
m_name="""m2"""
)
m3 = M.score_to_position.v4(
input_1=m2.data,
score_field="volume_ratio",
hold_count=5,
total_position=1,
position_expr="1 AS position",
m_name="""m3"""
)
m4 = M.extract_data_dai.v18(
sql=m3.data,
start_date="""2019-01-01""",
start_date_bound_to_trading_date=True,
end_date="""2025-01-01""",
end_date_bound_to_trading_date=True,
before_start_days=90,
keep_before=False,
debug=False,
m_name="""m4"""
)
m5 = M.bigtrader.v35(
data=m4.data,
initialize=m5_initialize_bigquant_run,
handle_data=m5_handle_data_bigquant_run,
capital_base=1000000,
frequency="daily",
product_type="股票",
rebalance_period_type="交易日",
rebalance_period_days="1",
order_price_field_buy="open",
order_price_field_sell="open",
benchmark="中证1000指数",
plot_charts=True,
backtest_only=False,
m_name="m5"
)
# </aistudiograph>
\