ddrwode/lm_code
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master'

Browse Source
This commit is contained in:
Administrator
2026-01-28 18:17:33 +08:00
parent 7a52508734 70aa29c820
commit eb237a0d56
5 changed files with 169 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
1111
View File

@@ -47,3 +47,7 @@ RSI超买超卖
申请API操作频率：预计API的调用频率将根据市场波动而变化，通常情况下，我的API请求频率大约在每秒5-10次，有时会根据策略的调整需要进行更频繁的请求。
申请API大概体量：预计每日API请求量为5,000 - 15,000次，具体取决于市场波动和交易策略的复杂度。月度交易量预估为50,000 - 200,000 ETH/BTC。
申请API对接后预估增长目标：通过API对接，我预计能够实现量化策略的高效执行和自动化交易，减少人工干预和错误。目标是在对接后的3个月内，将交易频率提高50%，并将月盈利提高20%至30%。
45.84
4998.6

143
bitmart/回测-三分之一策略.py
View File

@@ -4,10 +4,8 @@
========== 策略规则 ==========
1. 触发价格计算（基于有效的前一根K线，实体>=0.1）：
- 实体上边界 = max(开盘价, 收盘价)
- 实体下边界 = min(开盘价, 收盘价)
- 做多触发价格 = 实体上边界 - 实体/3（进入实体上部1/3区域做多）
- 做空触发价格 = 实体下边界 + 实体/3（进入实体下部1/3区域做空）
- 做多触发价格 = 收盘价 + 实体/3（从收盘价往上涨1/3）
- 做空触发价格 = 收盘价 - 实体/3（从收盘价往下跌1/3）
2. 信号触发条件：
- 当前K线最高价 >= 做多触发价格 → 做多信号
@@ -16,7 +14,7 @@
3. 执行逻辑：
- 做多时遇到做空信号 -> 平多并反手开空
- 做空时遇到做多信号 -> 平空并反手开多
- 如果同时触发两个方向，以距离开盘价更近的方向优先
- 同一根K线内只交易一次，防止频繁反手
4. 实体过滤：
- 如果前一根K线的实体部分（|open - close|）< 0.1，继续往前查找
@@ -24,15 +22,13 @@
示例1（阳线）：
前一根K线：开盘3000，收盘3100（阳线，实体=100）
- 实体上边界 = 3100，实体下边界 = 3000
- 做多触发价格 = 3100 - 33.33 = 3066.67（价格涨到这里做多）
- 做空触发价格 = 3000 + 33.33 = 3033.33（价格跌到这里做空）
- 做多触发价格 = 3100 + 33 = 3133（继续上涨做多）
- 做空触发价格 = 3100 - 33 = 3067（回调做空）
示例2（阴线）：
前一根K线：开盘3100，收盘3000（阴线，实体=100）
- 实体上边界 = 3100，实体下边界 = 3000
- 做多触发价格 = 3100 - 33.33 = 3066.67（价格涨到这里做多）
- 做空触发价格 = 3000 + 33.33 = 3033.33（价格跌到这里做空）
- 做多触发价格 = 3000 + 33 = 3033（反弹做多）
- 做空触发价格 = 3000 - 33 = 2967（继续下跌做空）
"""
import datetime
@@ -127,23 +123,30 @@ def get_one_third_levels(prev):
计算前一根K线实体的 1/3 双向触发价格
返回：(做多触发价格, 做空触发价格)
基于实体的上下边界计算：
- 做多触发价格 = 实体上边界 - 实体/3（进入实体上部1/3区域做多）
- 做空触发价格 = 实体下边界 + 实体/3（进入实体下部1/3区域做空）
基于收盘价计算（无论阴线阳线）：
- 做多触发价格 = 收盘价 + 实体/3（从收盘价往上涨1/3实体）
- 做空触发价格 = 收盘价 - 实体/3（从收盘价往下跌1/3实体）
示例：
阳线 open=3000, close=3100, 实体=100
- 做多触发 = 3100 + 33 = 3133（继续涨）
- 做空触发 = 3100 - 33 = 3067（回调）
阴线 open=3100, close=3000, 实体=100
- 做多触发 = 3000 + 33 = 3033（反弹）
- 做空触发 = 3000 - 33 = 2967（继续跌）
"""
p_open = float(prev['open'])
p_close = float(prev['close'])
body_high = max(p_open, p_close) # 实体上边界
body_low = min(p_open, p_close) # 实体下边界
body = body_high - body_low
body = abs(p_open - p_close)
if body < 0.001: # 十字星，忽略
return None, None
# 基于实体边界的双向触发价格
long_trigger = body_high - body / 3 # 进入实体上部1/3区域触发做多
short_trigger = body_low + body / 3 # 进入实体下部1/3区域触发做空
# 基于收盘价的双向触发价格
long_trigger = p_close + body / 3 # 从收盘价往上涨1/3触发做多
short_trigger = p_close - body / 3 # 从收盘价往下跌1/3触发做空
return long_trigger, short_trigger
@@ -225,7 +228,7 @@ def get_data_by_date(model, date_str: str):
# ========================= 回测逻辑 =========================
def backtest_one_third_strategy(dates: List[str]):
"""三分之一回归策略回测"""
"""三分之一回归策略回测（优化版）"""
all_data: List[Dict] = []
for d in dates:
@@ -258,23 +261,40 @@ def backtest_one_third_strategy(dates: List[str]):
'long': {'count': 0, 'wins': 0, 'total_profit': 0.0, 'name': '做多'},
'short': {'count': 0, 'wins': 0, 'total_profit': 0.0, 'name': '做空'},
}
# 额外统计信息
extra_stats = {
'same_dir_ignored': 0, # 同向信号被忽略次数
'no_signal_bars': 0, # 无信号K线数
'total_bars': len(all_data) - 1, # 总K线数（排除第一根）
}
trades: List[Dict] = []
current_position: Optional[Dict] = None
last_trade_bar: Optional[int] = None # 记录上次交易的K线索引，防止同一K线重复交易
for idx in range(1, len(all_data)):
curr = all_data[idx]
# 使用新的check_trigger函数，它会自动查找实体>=0.1的前一根K线
# 使用check_trigger函数，它会自动查找实体>=0.1的前一根K线
direction, trigger_price, valid_prev_idx = check_trigger(all_data, idx, min_body_size=0.1)
# 获取有效的前一根K线用于日志输出
valid_prev = all_data[valid_prev_idx] if valid_prev_idx is not None else None
# 无信号时跳过
if direction is None:
extra_stats['no_signal_bars'] += 1
continue
# 同一K线内已交易，跳过（与交易代码逻辑一致）
if last_trade_bar == idx:
continue
# 空仓时，有信号就开仓
if current_position is None:
if direction and valid_prev is not None:
# 调试：打印开仓时的K线信息
if valid_prev is not None:
# 打印开仓时的K线信息
prev_time = datetime.datetime.fromtimestamp(valid_prev['id'] / 1000).strftime('%Y-%m-%d %H:%M')
curr_time = datetime.datetime.fromtimestamp(curr['id'] / 1000).strftime('%Y-%m-%d %H:%M')
prev_type = "阳线" if is_bullish(valid_prev) else ("阴线" if is_bearish(valid_prev) else "十字星")
@@ -291,13 +311,19 @@ def backtest_one_third_strategy(dates: List[str]):
'entry_bar': idx
}
stats[direction]['count'] += 1
last_trade_bar = idx # 记录交易K线
continue
# 有仓位时，检查是否有反向信号
# 有仓位时，检查信号
pos_dir = current_position['direction']
if direction and direction != pos_dir and valid_prev is not None:
# 平仓
# 同向信号，忽略（与交易代码逻辑一致）
if direction == pos_dir:
extra_stats['same_dir_ignored'] += 1
continue
# 反向信号，平仓反手
if valid_prev is not None:
exit_price = trigger_price
if pos_dir == 'long':
@@ -305,7 +331,7 @@ def backtest_one_third_strategy(dates: List[str]):
else:
diff = current_position['entry_price'] - exit_price
# 调试：打印平仓时的K线信息
# 打印平仓时的K线信息
prev_time = datetime.datetime.fromtimestamp(valid_prev['id'] / 1000).strftime('%Y-%m-%d %H:%M')
curr_time = datetime.datetime.fromtimestamp(curr['id'] / 1000).strftime('%Y-%m-%d %H:%M')
prev_type = "阳线" if is_bullish(valid_prev) else ("阴线" if is_bearish(valid_prev) else "十字星")
@@ -323,7 +349,8 @@ def backtest_one_third_strategy(dates: List[str]):
'direction': '做多' if pos_dir == 'long' else '做空',
'entry': current_position['entry_price'],
'exit': exit_price,
'diff': diff
'diff': diff,
'hold_bars': idx - current_position['entry_bar'] # 持仓K线数
})
stats[pos_dir]['total_profit'] += diff
@@ -338,10 +365,12 @@ def backtest_one_third_strategy(dates: List[str]):
'entry_bar': idx
}
stats[direction]['count'] += 1
last_trade_bar = idx # 记录交易K线
# 尾仓处理
if current_position:
last = all_data[-1]
last_idx = len(all_data) - 1
exit_price = float(last['close'])
pos_dir = current_position['direction']
@@ -358,13 +387,23 @@ def backtest_one_third_strategy(dates: List[str]):
'direction': '做多' if pos_dir == 'long' else '做空',
'entry': current_position['entry_price'],
'exit': exit_price,
'diff': diff
'diff': diff,
'hold_bars': last_idx - current_position['entry_bar'], # 持仓K线数
'is_tail': True # 标记为尾仓平仓
})
stats[pos_dir]['total_profit'] += diff
if diff > 0:
stats[pos_dir]['wins'] += 1
logger.info(f"【尾仓平仓】{pos_dir} @ {exit_price:.2f}, 盈亏: {diff:.2f}")
return trades, stats, all_data
# 打印额外统计信息
logger.info(f"\n===== 信号统计 =====")
logger.info(f"总K线数: {extra_stats['total_bars']}")
logger.info(f"无信号K线: {extra_stats['no_signal_bars']} ({extra_stats['no_signal_bars']/extra_stats['total_bars']*100:.1f}%)")
logger.info(f"同向信号忽略: {extra_stats['same_dir_ignored']}")
return trades, stats, all_data, extra_stats
# ========================= 绘图函数 =========================
@@ -602,7 +641,7 @@ if __name__ == '__main__':
logger.info(f"查询日期范围：{START_DATE} 到 {END_DATE}，共 {len(dates)} 天")
# ==================== 执行回测 ====================
trades, stats, all_data = backtest_one_third_strategy(dates)
trades, stats, all_data, extra_stats = backtest_one_third_strategy(dates)
# ==================== 输出结果 ====================
logger.info("===== 每笔交易详情 =====")
@@ -645,14 +684,49 @@ if __name__ == '__main__':
# ==================== 汇总统计 ====================
total_net_profit = total_money_profit - total_fee
# 计算额外统计
win_count = len([t for t in trades if t['diff'] > 0])
lose_count = len([t for t in trades if t['diff'] <= 0])
total_win_rate = (win_count / len(trades) * 100) if trades else 0
# 计算平均持仓K线数
hold_bars_list = [t.get('hold_bars', 0) for t in trades if 'hold_bars' in t]
avg_hold_bars = sum(hold_bars_list) / len(hold_bars_list) if hold_bars_list else 0
# 计算最大连续亏损
max_consecutive_loss = 0
current_consecutive_loss = 0
for t in trades:
if t['diff'] <= 0:
current_consecutive_loss += 1
max_consecutive_loss = max(max_consecutive_loss, current_consecutive_loss)
else:
current_consecutive_loss = 0
# 计算最大回撤
cumulative_profit = 0
peak = 0
max_drawdown = 0
for t in trades:
cumulative_profit += t.get('net_profit', t['diff'])
peak = max(peak, cumulative_profit)
drawdown = peak - cumulative_profit
max_drawdown = max(max_drawdown, drawdown)
print(f"\n{'='*60}")
print(f"【三分之一回归策略 回测结果】")
print(f"{'='*60}")
print(f"交易笔数：{len(trades)}")
print(f"盈利笔数：{win_count} 亏损笔数：{lose_count}")
print(f"总胜率：{total_win_rate:.2f}%")
print(f"平均持仓K线数：{avg_hold_bars:.1f}")
print(f"最大连续亏损：{max_consecutive_loss} 笔")
print(f"{'='*60}")
print(f"总点差：{total_points_profit:.2f}")
print(f"总原始盈利：{total_money_profit:.2f}")
print(f"总手续费：{total_fee:.2f}")
print(f"总净利润：{total_net_profit:.2f}")
print(f"最大回撤：{max_drawdown:.2f}")
print(f"{'='*60}")
print("\n===== 方向统计 =====")
@@ -663,6 +737,11 @@ if __name__ == '__main__':
win_rate = (wins / count * 100) if count > 0 else 0.0
avg_p = (total_p / count) if count > 0 else 0.0
print(f"{v['name']}: 次数={count} 胜率={win_rate:.2f}% 总价差={total_p:.2f} 平均价差={avg_p:.2f}")
print("\n===== 信号统计 =====")
print(f"总K线数: {extra_stats['total_bars']}")
print(f"无信号K线: {extra_stats['no_signal_bars']} ({extra_stats['no_signal_bars']/extra_stats['total_bars']*100:.1f}%)")
print(f"同向信号忽略: {extra_stats['same_dir_ignored']}")
# ==================== 绘制图表 ====================
if trades and all_data:

BIN
bitmart/回测图表.png
View File

Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 141 KiB

After

Width:  |  Height:  |  Size: 144 KiB

2
bitmart/回测图表_交互式.html
View File

File diff suppressed because one or more lines are too long

77
交易/bitmart-三分之一策略交易.py
View File

@@ -4,10 +4,8 @@ BitMart 三分之一回归策略交易（双向触发版）
策略规则：
1. 触发价格计算（基于有效的前一根K线，实体>=0.1）：
- 实体上边界 = max(开盘价, 收盘价)
- 实体下边界 = min(开盘价, 收盘价)
- 做多触发价格 = 实体上边界 - 实体/3（进入实体上部1/3区域做多）
- 做空触发价格 = 实体下边界 + 实体/3（进入实体下部1/3区域做空）
- 做多触发价格 = 收盘价 + 实体/3（从收盘价往上涨1/3）
- 做空触发价格 = 收盘价 - 实体/3（从收盘价往下跌1/3）
2. 信号触发条件：
- 当前K线最高价 >= 做多触发价格 → 做多信号
@@ -16,12 +14,17 @@ BitMart 三分之一回归策略交易（双向触发版）
3. 执行逻辑：
- 做多时遇到做空信号 -> 平多并反手开空
- 做空时遇到做多信号 -> 平空并反手开多
- 如果同时触发两个方向，以距离开盘价更近的方向优先
- 同一根K线内只交易一次，防止频繁反手
示例（阳线）：
示例1（阳线）：
前一根K线：开盘3000，收盘3100（阳线，实体=100）
- 做多触发价格 = 3100 - 33.33 = 3066.67（价格涨到这里做多）
- 做空触发价格 = 3000 + 33.33 = 3033.33（价格跌到这里做空）
- 做多触发价格 = 3100 + 33 = 3133（继续上涨做多）
- 做空触发价格 = 3100 - 33 = 3067（回调做空）←当前跌到这里就做空
示例2（阴线）：
前一根K线：开盘3100，收盘3000（阴线，实体=100）
- 做多触发价格 = 3000 + 33 = 3033（反弹做多）
- 做空触发价格 = 3000 - 33 = 2967（继续下跌做空）
"""
import time
@@ -75,6 +78,7 @@ class BitmartOneThirdStrategy:
self.check_interval = 3 # 检测间隔（秒）
self.last_trigger_kline_id = None # 记录上次触发信号的K线ID，避免同一K线重复触发
self.last_trigger_direction = None # 记录上次触发的方向
self.last_trade_kline_id = None # 记录上次实际交易的K线ID，防止同一K线内频繁反手
# ========================= 三分之一策略核心函数 =========================
@@ -111,23 +115,30 @@ class BitmartOneThirdStrategy:
计算前一根K线实体的 1/3 双向触发价格
返回：(做多触发价格, 做空触发价格)
基于实体的上下边界计算：
- 做多触发价格 = 实体上边界 - 实体/3（进入实体上部1/3区域做多）
- 做空触发价格 = 实体下边界 + 实体/3（进入实体下部1/3区域做空）
基于收盘价计算（无论阴线阳线）：
- 做多触发价格 = 收盘价 + 实体/3（从收盘价往上涨1/3实体）
- 做空触发价格 = 收盘价 - 实体/3（从收盘价往下跌1/3实体）
示例：
阳线 open=3000, close=3100, 实体=100
- 做多触发 = 3100 + 33 = 3133（继续涨）
- 做空触发 = 3100 - 33 = 3067（回调）
阴线 open=3100, close=3000, 实体=100
- 做多触发 = 3000 + 33 = 3033（反弹）
- 做空触发 = 3000 - 33 = 2967（继续跌）
"""
p_open = float(prev['open'])
p_close = float(prev['close'])
body_high = max(p_open, p_close) # 实体上边界
body_low = min(p_open, p_close) # 实体下边界
body = body_high - body_low
body = abs(p_open - p_close)
if body < 0.001: # 十字星，忽略
return None, None
# 基于实体边界的双向触发价格
long_trigger = body_high - body / 3 # 进入实体上部1/3区域触发做多
short_trigger = body_low + body / 3 # 进入实体下部1/3区域触发做空
# 基于收盘价的双向触发价格
long_trigger = p_close + body / 3 # 从收盘价往上涨1/3触发做多
short_trigger = p_close - body / 3 # 从收盘价往下跌1/3触发做空
return long_trigger, short_trigger
@@ -520,6 +531,17 @@ class BitmartOneThirdStrategy:
direction, trigger_price, valid_prev, curr_kline = self.check_realtime_trigger(kline_data)
if direction:
curr_kline_id = curr_kline['id']
# 检查是否在同一K线内已经交易过（防止频繁反手）
if self.last_trade_kline_id == curr_kline_id:
logger.debug(f"同一K线内已交易，跳过本次{direction}信号")
# 更新触发记录，避免重复日志
self.last_trigger_kline_id = curr_kline_id
self.last_trigger_direction = direction
time.sleep(self.check_interval)
continue
# 获取持仓状态
if not self.get_position_status():
logger.warning("获取仓位信息失败")
@@ -530,6 +552,14 @@ class BitmartOneThirdStrategy:
prev_type = "阳线" if self.is_bullish(valid_prev) else "阴线"
prev_body = self.get_body_size(valid_prev)
# 检查信号与持仓是否同向（避免重复日志）
if (direction == "long" and self.start == 1) or (direction == "short" and self.start == -1):
# 信号与持仓同向，静默忽略
self.last_trigger_kline_id = curr_kline_id
self.last_trigger_direction = direction
time.sleep(self.check_interval)
continue
logger.info(f"{'='*50}")
logger.info(f"🚨 检测到{direction}信号！触发价格: {trigger_price:.2f}")
logger.info(f" 有效前一根[{prev_time}]: {prev_type} 实体={prev_body:.2f} O={valid_prev['open']:.2f} C={valid_prev['close']:.2f}")
@@ -554,8 +584,6 @@ class BitmartOneThirdStrategy:
logger.info("📈 无仓位，开多")
self.开单(marketPriceLongOrder=1, size=trade_size)
executed = True
else:
logger.info("已有多仓，忽略做多信号")
elif direction == "short":
if self.start == 1: # 当前多仓，平多开空
@@ -568,13 +596,14 @@ class BitmartOneThirdStrategy:
logger.info("📉 无仓位，开空")
self.开单(marketPriceLongOrder=-1, size=trade_size)
executed = True
else:
logger.info("已有空仓，忽略做空信号")
# 记录本次触发，避免同一K线重复触发
# 记录本次触发
self.last_trigger_kline_id = curr_kline_id
self.last_trigger_direction = direction
if executed:
self.last_trigger_kline_id = curr_kline['id']
self.last_trigger_direction = direction
# 记录交易K线，防止同一K线内频繁反手
self.last_trade_kline_id = curr_kline_id
# 交易后立即发送持仓信息
self.get_position_status()
self._send_position_message(curr_kline)