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

加入精准回测数据

Browse Source
This commit is contained in:
27942
2026-02-01 22:29:51 +08:00
parent f25c725eda
commit 8bc54a3bb4
2 changed files with 85493 additions and 50560 deletions
Download Patch File Download Diff File Expand all files Collapse all files

135975
bitmart/backtest_one_third_trades.csv
View File

File diff suppressed because it is too large Load Diff

78
bitmart/回测-三分之一策略-精准版.py
View File

@@ -1,11 +1,11 @@
"""
BitMart 三分之一回归策略回测（精准版）
使用5分钟K线周期计算触发价格，1分钟K线判断触发顺序
BitMart 四分之一回归策略回测（精准版）
使用3分钟K线周期计算触发价格，1分钟K线判断触发顺序
========== 策略规则 ==========
1. 触发价格计算（基于有效的前一根K线，实体>=0.1）：
- 做多触发价格 = 收盘价 + 实体/3（从收盘价往上涨1/3）
- 做空触发价格 = 收盘价 - 实体/3（从收盘价往下跌1/3）
- 做多触发价格 = 收盘价 + 实体/4（从收盘价往上涨1/4）
- 做空触发价格 = 收盘价 - 实体/4（从收盘价往下跌1/4）
2. 信号触发条件：
- 当前K线最高价 >= 做多触发价格 → 做多信号
@@ -14,11 +14,11 @@ BitMart 三分之一回归策略回测（精准版）
3. 执行逻辑：
- 做多时遇到做空信号 -> 平多并反手开空
- 做空时遇到做多信号 -> 平空并反手开多
- 同一根5分钟K线内只交易一次
- 同一根3分钟K线内只交易一次
4. 精准判断（使用1分钟K线）：
- 当一根5分钟K线同时触及做多和做空价格时
- 使用该5分钟K线对应的5根1分钟K线来判断哪个方向先被触发
- 当一根3分钟K线同时触及做多和做空价格时
- 使用该3分钟K线对应的3根1分钟K线来判断哪个方向先被触发
- 这样可以更精准地还原真实交易场景
"""
@@ -49,8 +49,8 @@ class BitMartETH1m(Model):
table_name = 'bitmart_eth_1m'
class BitMartETH5m(Model):
"""5分钟K线模型"""
class BitMartETH3m(Model):
"""3分钟K线模型"""
id = BigIntegerField(primary_key=True) # 时间戳（毫秒级）
open = FloatField(null=True)
high = FloatField(null=True)
@@ -59,7 +59,7 @@ class BitMartETH5m(Model):
class Meta:
database = db
table_name = 'bitmart_eth_5m'
table_name = 'bitmart_eth_3m'
# 连接数据库
@@ -100,14 +100,14 @@ def find_valid_prev_bar(all_data, current_idx, min_body_size=0.1):
return None, None
def get_one_third_levels(prev):
def get_one_fourth_levels(prev):
"""
计算前一根K线实体的 1/3 双向触发价格
计算前一根K线实体的 1/4 双向触发价格
返回：(做多触发价格, 做空触发价格)
基于收盘价计算（无论阴线阳线）：
- 做多触发价格 = 收盘价 + 实体/3（从收盘价往上涨1/3实体）
- 做空触发价格 = 收盘价 - 实体/3（从收盘价往下跌1/3实体）
- 做多触发价格 = 收盘价 + 实体/4（从收盘价往上涨1/4实体）
- 做空触发价格 = 收盘价 - 实体/4（从收盘价往下跌1/4实体）
"""
p_open = float(prev['open'])
p_close = float(prev['close'])
@@ -118,14 +118,14 @@ def get_one_third_levels(prev):
return None, None
# 基于收盘价的双向触发价格
long_trigger = p_close + body / 3
short_trigger = p_close - body / 3
long_trigger = p_close + body / 4
short_trigger = p_close - body / 4
return long_trigger, short_trigger
def get_5m_data_by_date(date_str: str) -> List[Dict]:
"""按天获取5分钟K线数据"""
def get_3m_data_by_date(date_str: str) -> List[Dict]:
"""按天获取3分钟K线数据"""
try:
target_date = datetime.datetime.strptime(date_str, '%Y-%m-%d')
except ValueError:
@@ -135,9 +135,9 @@ def get_5m_data_by_date(date_str: str) -> List[Dict]:
start_ts = int(target_date.timestamp() * 1000)
end_ts = int((target_date + datetime.timedelta(days=1)).timestamp() * 1000) - 1
query = BitMartETH5m.select().where(
BitMartETH5m.id.between(start_ts, end_ts)
).order_by(BitMartETH5m.id.asc())
query = BitMartETH3m.select().where(
BitMartETH3m.id.between(start_ts, end_ts)
).order_by(BitMartETH3m.id.asc())
data = [{'id': i.id, 'open': i.open, 'high': i.high, 'low': i.low, 'close': i.close} for i in query]
return data
@@ -158,14 +158,14 @@ def get_1m_data_by_range(start_ts: int, end_ts: int) -> List[Dict]:
return data
def get_1m_data_for_5m_bar(bar_5m: Dict) -> List[Dict]:
def get_1m_data_for_3m_bar(bar_3m: Dict) -> List[Dict]:
"""
获取5分钟K线对应的5根1分钟K线
:param bar_5m: 5分钟K线数据
:return: 对应的1分钟K线数据列表（最多5根）
获取3分钟K线对应的3根1分钟K线
:param bar_3m: 3分钟K线数据
:return: 对应的1分钟K线数据列表（最多3根）
"""
start_ts = bar_5m['id']
end_ts = start_ts + 5 * 60 * 1000 # 5分钟后
start_ts = bar_3m['id']
end_ts = start_ts + 3 * 60 * 1000 # 3分钟后
return get_1m_data_by_range(start_ts, end_ts)
@@ -175,9 +175,9 @@ def determine_trigger_order_by_1m(
short_trigger: float
) -> str:
"""
使用1分钟K线精确判断在5分钟周期内，是先触发做多还是做空
使用1分钟K线精确判断在3分钟周期内，是先触发做多还是做空
:param bars_1m: 5根1分钟K线数据
:param bars_1m: 3根1分钟K线数据
:param long_trigger: 做多触发价格
:param short_trigger: 做空触发价格
:return: 'long', 'short', 或 None
@@ -216,12 +216,12 @@ def determine_trigger_order_by_1m(
def check_trigger_with_1m(
all_data_5m: List[Dict],
all_data_3m: List[Dict],
current_idx: int,
min_body_size: float = 0.1
) -> tuple:
"""
检查当前5分钟K线是否触发了交易信号
检查当前3分钟K线是否触发了交易信号
如果同时触发两个方向，使用1分钟K线精确判断顺序
返回：(方向, 触发价格, 有效前一根K线索引, 1分钟数据是否使用)
@@ -229,15 +229,15 @@ def check_trigger_with_1m(
if current_idx <= 0:
return None, None, None, False
curr = all_data_5m[current_idx]
curr = all_data_3m[current_idx]
# 查找实体>=min_body_size的前一根K线
valid_prev_idx, prev = find_valid_prev_bar(all_data_5m, current_idx, min_body_size)
valid_prev_idx, prev = find_valid_prev_bar(all_data_3m, current_idx, min_body_size)
if prev is None:
return None, None, None, False
long_trigger, short_trigger = get_one_third_levels(prev)
long_trigger, short_trigger = get_one_fourth_levels(prev)
if long_trigger is None:
return None, None, None, False
@@ -251,7 +251,7 @@ def check_trigger_with_1m(
# 如果两个方向都触发，使用1分钟K线精确判断
if long_triggered and short_triggered:
bars_1m = get_1m_data_for_5m_bar(curr)
bars_1m = get_1m_data_for_3m_bar(curr)
if bars_1m:
direction = determine_trigger_order_by_1m(bars_1m, long_trigger, short_trigger)
@@ -287,16 +287,16 @@ def backtest_one_third_strategy(dates: List[str], min_body_size: float = 0.1):
:param min_body_size: 最小实体大小
:return: (trades, stats)
"""
# 获取所有5分钟K线数据
# 获取所有3分钟K线数据
all_data: List[Dict] = []
total_queried = 0
for d in dates:
day_data = get_5m_data_by_date(d)
day_data = get_3m_data_by_date(d)
all_data.extend(day_data)
if day_data:
total_queried += len(day_data)
logger.info(f"总共查询了 {len(dates)} 天，获取到 {total_queried} 条5分钟K线数据")
logger.info(f"总共查询了 {len(dates)} 天，获取到 {total_queried} 条3分钟K线数据")
if not all_data:
logger.warning("未获取到任何数据，请检查数据库")
@@ -507,7 +507,7 @@ if __name__ == '__main__':
total_net_profit = total_money_profit - total_fee
print(f"\n{'='*60}")
print(f"【BitMart 三分之一策略回测结果（5分钟K线 + 1分钟精准判断）】")
print(f"【BitMart 四分之一策略回测结果（3分钟K线 + 1分钟精准判断）】")
print(f"{'='*60}")
print(f"回测周期：{START_DATE} 到 {END_DATE}")
print(f"最小实体要求：{MIN_BODY_SIZE}")