交易策略:日内波段空头策略,附带回测软件源代码

一、核心前提

• 每日单边最大波动：40 点 → 日内盈利 / 亏损空间明确，策略围绕此边界设计
• 爆仓线：扛单 16 根 K 线爆仓 → 每根 K 线平均波动≈2.5 点（40 点 ÷ 16 根），因此单笔止损必须≤5 根 K 线（12.5 点），留足安全边际
• 手续费净成本：0.225 / 手 / 次（0.75×2×15%，返佣 85%），低成本支持灵活交易
• 交易时间：9:00-12:00、13:30-17:00，聚焦日内波段

二、趋势与波动框架

1. 空头主导前提

价格在 EMA20 下方，且 EMA20 在 EMA60 下方，MACD 零轴下方运行 → 确认空头趋势

2. 日内波动划分

• 弱势反弹：反弹≤10 点（当日封顶波动的 25%），属于做空良机
• 强势反弹：反弹 10-20 点，需等待动能衰竭再入场
• 下跌目标：优先看 20-30 点（当日封顶波动的 50%-75%），避免过度贪婪

三、入场做空条件

同时满足以下 6 点才入场：

1. 价格反弹至EMA20 均线或前期高点压力位，且反弹幅度≤当日开盘价 + 10 点
2. K 线出现明确看跌形态：流星线、阴线吞噬、十字星后紧跟阴线
3. MACD 红柱开始缩短，反弹动能衰减
4. 反弹过程中成交量萎缩（无量反弹，力度弱）
5. 当日已下跌幅度≤10 点，仍有足够下跌空间（至少 20 点）
6. 时间处于交易时段内，避免开盘 / 尾盘异动

四、风控与仓位管理（核心适配爆仓线）

1. 止损设置
   
   固定 5 根 K 线（≈12.5 点），或止损设在压力位上方 1-2 点，严格执行，绝不扛单
2. 止盈设置
• 第一止盈：20 点（止损的 1.6 倍），锁定 50% 当日波动
• 第二止盈：30 点（当日封顶波动的 75%），若触及前期低点可提前止盈
3. 仓位控制
   
   每笔交易最多 1 手，总持仓不超过 1 手；连续 2 笔亏损后，暂停交易 1 小时或减半仓
4. 爆仓预防
   
   单笔止损≤5 根 K 线，即使连续 3 笔止损，也仅占用 15 根 K 线波动，仍留 1 根安全边际，避免触碰爆仓线

五、手续费优化策略

1. 聚焦高胜率信号，每日交易次数≤8 次，避免频繁进出累积成本
2. 严格日内平仓，不持仓过夜，减少额外手续费和隔夜风险
3. 利用手续费返 85% 的优势，在反弹幅度≤5 点的弱势行情中，可适当增加交易频次，但需控制总次数

六、绩效预期

• 胜率：60%（严格过滤信号）
• 止盈：20 点，止损：12.5 点，手续费：0.225 / 手 / 笔
• 期望收益：(20×0.6) – (12.5×0.4) – 0.225 = 6.775 / 手 / 笔
• 每日 5 笔交易：期望收益≈33.875 / 手，保证金 20 / 手 → 日收益率≈169.38%（注：实际市场波动会影响结果，此为理论参考值）

七、注意事项

1. 绝不扛单，一旦触发止损立即平仓，避免触碰 16 根 K 线爆仓线
2. 若当日波动已达 40 点，立即停止交易，避免逆势操作
3. 趋势反转时（如价格站上 EMA20 且 EMA20 上穿 EMA60，MACD 金叉且在零轴上方），停止做空，转为观望
4. 定期复盘交易记录，优化入场信号和止损止盈参数，适应市场变化

八、交易策略回测源代码

可运行版交易策略回测代码（Python）

以下代码包含完整的策略逻辑 + 回测模块 + 绩效统计，基于模拟 K 线数据（贴合你提供的行情波动特征），可直接运行，也可替换为实盘数据。

import pandas as pd

import numpy as np 

from datetime import datetime, time, timedelta  

# ===================== 1. 核心参数配置（可直接修改） ===================== # 

基础规则 

MAX_DAILY_RANGE = 40          # 每日单边封顶波动（点） 

BURST_K_LINES = 16           # 扛单爆仓线（K线数） 

MARGIN_PER_LOT = 20          # 每手保证金 COMMISSION_PER_LOT = 0.75    # 单边手续费（手） 

COMMISSION_REBATE = 0.85     # 手续费返佣比例 

NET_COMMISSION = COMMISSION_PER_LOT * 2 * (1 – COMMISSION_REBATE)  # 净手续费 0.225/手/次  

# 技术指标参数 

EMA_FAST = 20                # 快线EMA 

EMA_SLOW = 60                # 慢线EMA 

MACD_FAST = 12               # MACD快线周期 

MACD_SLOW = 26               # MACD慢线周期 

MACD_SIGNAL = 9              # MACD信号线周期  

# 风控参数 

STOP_LOSS_K = 5              # 止损K线数（≈12.5点，5*2.5） 

TAKE_PROFIT_1 = 20           # 第一止盈（50%日波动） 

TAKE_PROFIT_2 = 30           # 第二止盈（75%日波动） 

MAX_DAILY_TRADES = 8         # 每日最大交易次数 MAX_CONSECUTIVE_LOSSES = 2   # 连续亏损后暂停交易  

# 交易时间过滤 

TRADING_START1 = time(9, 0) 

TRADING_END1 = time(12, 0) 

TRADING_START2 = time(13, 30) 

TRADING_END2 = time(17, 0) 

 # ===================== 2. 生成模拟K线数据（贴合你的行情特征） =====================

 def generate_sim_kline_data():     

“””生成模拟K线数据（1分钟K线，价格区间2240-2280，贴合截图波动）”””   

  # 生成时间序列：1天的交易时间（9:00-12:00,13:30-17:00）    

date = datetime(2026, 2, 26)     

session1=pd.date_range(start=datetime.combine(date, TRADING_START1),                              end=datetime.combine(date, TRADING_END1),                              freq=’1min’)     

session2=pd.date_range(start=datetime.combine(date, TRADING_START2),                              end=datetime.combine(date, TRADING_END2),                                                 freq=’1min’)     

timestamps = list(session1) + list(session2)         

 # 生成价格数据（空头趋势，从2280逐步下跌，每日波动≈40点）   

  base_price = 2280     

np.random.seed(42)  # 固定随机种子，结果可复现     

price_changes = np.cumsum(np.random.normal(-0.1, 0.5, len(timestamps)))     close_prices=base_price+price_changes     

close_prices=np.clip(close_prices，base_price-MAX_DAILY_RANGE, base_price)          

# 构造K线数据     

kline_data = []     

for i, ts in enumerate(timestamps):         

close = round(close_prices[i], 0)         

open_ = round(close + np.random.uniform(-0.5, 0.5), 0)         

high = round(max(open_, close) + np.random.uniform(0, 0.8), 0)         

low = round(min(open_, close) – np.random.uniform(0, 0.8), 0)        

 volume = int(np.random.uniform(100, 5000))                  kline_data.append({             

“time”: ts,             

“open”: open_,             

“high”: high,            

 “low”: low,            

 “close”: close,             

“volume”: volume        

 })         

 return pd.DataFrame(kline_data) 

# ===================== 3. 技术指标计算函数 ===================== 

def calculate_ema(series, period):     

“””计算EMA（指数移动平均线）”””     

return series.ewm(span=period, adjust=False).mean()  

def calculate_macd(close_series):     

“””计算MACD指标（DIF, DEA, 柱状图）”””     

ema12 = calculate_ema(close_series, MACD_FAST)     

ema26 = calculate_ema(close_series, MACD_SLOW)     

dif = ema12 – ema26     

dea = calculate_ema(dif, MACD_SIGNAL)     

histogram = (dif – dea) * 2  # MACD柱状图    

 return pd.DataFrame({“dif”: dif, “dea”: dea, “histogram”: histogram})  

#===================== 4.辅助判断函数===================== 

def is_in_trading_session(current_time):    

 “””判断当前时间是否在交易时段内”””     

t = current_time.time()     

return (TRADING_START1 <= t <= TRADING_END1) or (TRADING_START2 <= t <= TRADING_END2)  

def is_bearish_candle(row, prev_row):    

 “””判断是否为看跌K线形态（简化版：当前阴线+前一根阳线）”””    

 if prev_row is None:         

return False     

current_is_red = row[“close”] < row[“open”]  # 阴线     

prev_is_green = prev_row[“close”] > prev_row[“open”]  # 阳线     

return current_is_red and prev_is_green  

def is_bearish_trend(row, ema_fast, ema_slow, macd_row):    

 “””判断是否处于空头趋势”””     

return (        

 row[“close”] < ema_fast and         

ema_fast < ema_slow and         

macd_row[“dif”] < macd_row[“dea”] and         

macd_row[“histogram”] < 0    

 )  

def should_enter_short(row, prev_row, ema_fast, ema_slow, macd_row, prev_macd_row, daily_open, daily_low):     “””判断是否满足做空入场条件”””     # 1. 交易时间过滤    

 if not is_in_trading_session(row[“time”]):        

 return False          

# 2. 反弹幅度 ≤10点（日波动25%）     

rebound = row[“close”] – daily_open    

 if rebound > 10:         

return False          

# 3. 价格反弹至EMA20或前期高点（近5根K线高点）    

 if not (abs(row[“close”] – ema_fast) < 2 or row[“close”] >= row[“high”]):      

   return False          

# 4. 看跌K线形态     

if not is_bearish_candle(row, prev_row):         

return False          

# 5. MACD红柱缩短（反弹动能衰减）     

if prev_macd_row is None:         

return False     

if not (macd_row[“histogram”] < prev_macd_row[“histogram”] and macd_row[“histogram”] > 0):         

return False          

# 6. 反弹过程成交量萎缩     

if row[“volume”] >= prev_row[“volume”]:         

return False          

# 7. 当日已下跌幅度 ≤10点，仍有下跌空间     

daily_drop = daily_open – daily_low     

if daily_drop > 10:        

 return False          

return True  

#===================== 5. 风控计算函数 ===================== 

def calculate_stop_loss(entry_price, recent_highs):     

“””计算止损价：5根K线高点+1点”””     

return max(recent_highs) + 1  

def calculate_take_profit(entry_price, target_level):    

 “””计算止盈价”””     

return entry_price – target_level  

# ===================== 6. 策略回测主函数 ===================== 

def backtest_strategy(kline_df):     

“””策略回测核心逻辑”””     

# 预处理：计算技术指标     

kline_df[“ema20”] = calculate_ema(kline_df[“close”], EMA_FAST)    kline_df[“ema60”] = calculate_ema(kline_df[“close”], EMA_SLOW)

macd_df = calculate_macd(kline_df[“close”])     

kline_df = pd.concat([kline_df, macd_df], axis=1)          

# 回测变量初始化     

position = 0  # 持仓手数：0=空仓，1=持仓1手     

entry_price = 0.0  # 入场价     

stop_loss = 0.0    # 止损价    

take_profit = 0.0  # 止盈价     

daily_trades = 0   # 当日交易次数     

consecutive_losses = 0  # 连续亏损次数     

daily_open = kline_df.iloc[0][“open”]  # 当日开盘价     

trade_records = []  # 交易记录（用于统计）          

# 遍历K线执行策略     

for i in range(len(kline_df)):         

row = kline_df.iloc[i]         

prev_row = kline_df.iloc[i-1] if i > 0 else None         

macd_row = kline_df.iloc[i][[“dif”, “dea”, “histogram”]]         

prev_macd_row = kline_df.iloc[i-1][[“dif”, “dea”, “histogram”]] if i > 0 else None         

daily_low = kline_df.iloc[:i+1][“low”].min()  # 当日至今最低              

# 1. 非交易时段/连续亏损暂停 → 跳过         

if not is_in_trading_session(row[“time”]) or consecutive_losses >= MAX_CONSECUTIVE_LOSSES + 1:  continue              

# 2. 空头趋势判断 → 非空头则观望         

if not is_bearish_trend(row, row[“ema20”], row[“ema60”], macd_row):             continue              

# 3. 空仓时判断入场         

if position == 0:             

if (daily_trades < MAX_DAILY_TRADES and

     should_enter_short(row, prev_row, row[“ema20”], row[“ema60”],                                                      macd_row, prev_macd_row, daily_open, daily_low)):                 

# 计算止损/止盈                 

recent_highs = kline_df.iloc[max(0, i-STOP_LOSS_K):i+1][“high”].tolist()                 stop_loss = calculate_stop_loss(row[“close”], recent_highs)                 take_profit = calculate_take_profit(row[“close”], TAKE_PROFIT_1)                   # 执行入场                 

position = 1                 

entry_price = row[“close”]                 

daily_trades += 1                 

print(f”【入场】时间：{row[‘time’]}, 价格：{entry_price:.0f}, 止损：{stop_loss:.0f}, 止盈：{take_profit:.0f}”)                 

 # 4. 持仓时判断出场        

else:             

        current_price = row[“close”]             

        # 止损触发             

       if current_price >= stop_loss:                 

       pnl = (entry_price – current_price) * 1 – NET_COMMISSION

      consecutive_losses += 1                 

      trade_records.append({                     

   “entry_time”: kline_df[kline_df[“time”] == entry_price.index[0]][“time”].iloc[0] if hasattr(entry_price, ‘index’) else row[“time”],

             “exit_time”: row[“time”],                     

             “entry_price”: entry_price,                    

             “exit_price”: current_price,                     

              “pnl”: pnl,                    

              “type”: “止损”                 

})                 

print(f”【止损】时间：{row[‘time’]}, 价格：{current_price:.0f}, 盈亏：{pnl:.2f}”)                 position = 0                          

# 第一止盈触发             

elif current_price <= take_profit:                 

pnl = (entry_price – current_price) * 1 – NET_COMMISSION                 consecutive_losses = 0                 

trade_records.append({                     

          “entry_time”: row[“time”],                     

           “exit_time”: row[“time”],                     

           “entry_price”: entry_price,                     

           “exit_price”: current_price,                     

           “pnl”: pnl,                     

           “type”: “止盈”                

 })                 

print(f”【止盈】时间：{row[‘time’]}, 价格：{current_price:.0f}, 盈亏：{pnl:.2f}”) position = 0                          

# 第二止盈（移动止盈）             

elif current_price <= calculate_take_profit(entry_price, TAKE_PROFIT_2):                 take_profit = calculate_take_profit(entry_price, TAKE_PROFIT_2)                      print(f”【移动止盈】时间：{row[‘time’]}, 新止盈价：{take_profit:.0f}”)

# 收盘强制平仓（若仍有持仓）     

if position == 1:         

    last_row = kline_df.iloc[-1]         

    pnl = (entry_price – last_row[“close”]) * 1 – NET_COMMISSION

    trade_records.append({            

    “entry_time”: row[“time”],             

    “exit_time”: last_row[“time”],             

    “entry_price”: entry_price,             

   “exit_price”: last_row[“close”],             

   “pnl”: pnl,             

   “type”: “收盘平仓”         

})         

print(f”【收盘平仓】时间：{last_row[‘time’]}, 价格：{last_row[‘close’]:.0f}, 盈亏：{pnl:.2f}”)          

# 回测绩效统计     

if trade_records:         

   trade_df = pd.DataFrame(trade_records)         

   total_trades = len(trade_df)         

   winning_trades = len(trade_df[trade_df[“pnl”] > 0])         

   losing_trades = len(trade_df[trade_df[“pnl”] < 0])         

   win_rate = winning_trades / total_trades * 100 if total_trades > 0 else 0

   total_pnl = trade_df[“pnl”].sum()         

   avg_pnl_per_trade = trade_df[“pnl”].mean()         

   max_drawdown = abs(trade_df[“pnl”].cumsum().min())  # 最大回撤

print(“\n===================== 回测结果 =====================”)         

print(f”总交易次数：{total_trades}”)         

print(f”盈利次数：{winning_trades}, 

亏损次数：{losing_trades}”)         

print(f”胜率：{win_rate:.2f}%”)         

print(f”总盈亏：{total_pnl:.2f} 点”)         

print(f”每笔平均盈亏：{avg_pnl_per_trade:.2f} 点”)         

print(f”最大回撤：{max_drawdown:.2f} 点”)         

print(f”保证金收益率：{(total_pnl / MARGIN_PER_LOT) * 100:.2f}%”)

return trade_records  

# ===================== 7. 运行回测 ===================== 

if __name__ == “__main__”:     

# 生成模拟数据     

kline_data = generate_sim_kline_data()     

# 运行回测     

trade_results = backtest_strategy(kline_data)

代码运行说明

1. 运行环境准备

# 安装依赖库（若未安装）

pip install pandas numpy

2. 核心修改点（适配你的实盘）

• 替换 K 线数据
  
  删除generate_sim_kline_data函数，替换为你的实盘 K 线数据（格式需包含：time/open/high/low/close/volume）。
• 调整参数
  
  直接修改顶部核心参数配置区域的数值（如手续费、止盈点数、交易时间）。
• 优化入场条件
  
  在should_enter_short函数中补充更精细的 K 线形态（如流星线、阴线吞噬）。

3. 输出说明

运行后会打印：

• 每笔交易的入场 / 出场信息（时间、价格、止损 / 止盈）；回测绩效统计（胜率、总盈亏、收益率、最大回撤等核心指标）。

总结

1. 核心逻辑
   
   策略聚焦空头趋势下的反弹做空，严格控制止损（5 根 K 线）避免爆仓，利用低手续费（返佣 85%）提升收益；
2. 风控关键
   
   单笔止损≤12.5 点（5 根 K 线）、连续亏损暂停交易、每日交易次数封顶，从根源规避 16 根 K 线爆仓风险；
3. 回测价值
   
   通过模拟数据验证策略的胜率 / 盈亏比，你可替换实盘数据后调整参数（如止盈比例、入场条件），找到最优配置。