233 lines
8.4 KiB
Python
233 lines
8.4 KiB
Python
|
|
# --- Do not remove these libs ---
|
|
from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter, CategoricalParameter
|
|
from typing import Dict, List
|
|
from functools import reduce
|
|
from pandas import DataFrame
|
|
import numpy as np
|
|
import pandas as pd
|
|
# --------------------------------
|
|
|
|
# 设置pandas选项以避免FutureWarning
|
|
pd.set_option('future.no_silent_downcasting', True)
|
|
|
|
import talib.abstract as ta
|
|
import freqtrade.vendor.qtpylib.indicators as qtpylib
|
|
from technical.util import resample_to_interval, resampled_merge
|
|
from freqtrade.persistence import Trade, Order
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional
|
|
import logging
|
|
logger = logging.getLogger(__name__)
|
|
# freqtrade plot-dataframe --strategy PatternTrader --datadir user_data/data/binance -c ./user_data/ChanLun_SOL_30.json --timerange=20250309-
|
|
|
|
# freqtrade trade -c ./user_data/Chan/config/ChanLun_BTC_30.json --strategy PatternTrader --strategy-path ./user_data/Chan/strategies
|
|
# freqtrade backtesting -c ./user_data/Chan/config/ChanLun_BTC_30.json --strategy PatternTrader --strategy-path ./user_data/Chan/strategies --timerange=20251023-
|
|
# freqtrade download-data -c ./user_data/Chan/config/ChanLun_BTC_30.json -t 1m --pairs BTC/USDT:USDT --timerange=20250501-
|
|
# freqtrade hyperopt --hyperopt-loss SharpeHyperOptLossDaily --spaces roi stoploss --strategy PatternTrader --strategy-path ./user_data/Chan/strategies -c ./user_data/Chan/config/ChanLun_BTC_30.json -e 200 --timerange=20250201-20250401
|
|
# freqtrade hyperopt --hyperopt-loss SharpeHyperOptLossDaily --spaces buy sell roi stoploss --strategy PatternTrader --strategy-path ./user_data/Chan/strategies -c ./user_data/Chan/config/ChanLun_BTC_30.json -e 600 --timerange=20250201-20250401
|
|
# sudo docker compose run --rm chan_btc backtesting -c ./user_data/Chan/config/ChanLun_BTC_30.json --strategy PatternTrader --strategy-path ./user_data/Chan/strategies --timerange=20250101-
|
|
# sudo docker compose run --rm chan_btc download-data -c ./user_data/Chan/config/ChanLun_BTC_30.json --pairs BTC/USDT:USDT -t 1m --timerange 20240101-
|
|
# sudo docker compose run --rm chan_btc trade -c ./user_data/Chan/config/ChanLun_BTC_30.json --strategy PatternTrader --strategy-path ./user_data/Chan/strategies
|
|
|
|
class PatternTrader(IStrategy):
|
|
"""
|
|
极简双均线策略
|
|
只使用双均线交叉作为唯一信号
|
|
"""
|
|
|
|
INTERFACE_VERSION: int = 3
|
|
|
|
# 极简参数
|
|
fast_ma: IntParameter = IntParameter(5, 15, default=8, space='buy') # 快速均线
|
|
slow_ma: IntParameter = IntParameter(20, 50, default=30, space='buy') # 慢速均线
|
|
|
|
# 添加一个简单的sell空间参数
|
|
exit_delay: IntParameter = IntParameter(1, 10, default=3, space='sell') # 出场延迟
|
|
|
|
# 时间框架
|
|
time: IntParameter = IntParameter(15, 60, default=30, space='buy')
|
|
|
|
# ROI 超参
|
|
roi_t1: IntParameter = IntParameter(10, 60, default=30, space='roi')
|
|
roi_t2: IntParameter = IntParameter(60, 240, default=120, space='roi')
|
|
roi_p1: DecimalParameter = DecimalParameter(0.02, 0.08, default=0.05, decimals=3, space='roi')
|
|
roi_p2: DecimalParameter = DecimalParameter(0.005, 0.03, default=0.01, decimals=3, space='roi')
|
|
|
|
# 合约交易参数
|
|
can_short = True
|
|
stoploss = -0.02 # 2% 止损
|
|
|
|
# 杠杆设置
|
|
lev: DecimalParameter = DecimalParameter(1.0, 3.0, default=2.0, decimals=1, space='buy')
|
|
|
|
# 运行设置
|
|
process_only_new_candles = False
|
|
startup_candle_count: int = 100
|
|
|
|
# ROI 外部覆盖
|
|
_roi_override: Optional[Dict[str, float]] = None
|
|
|
|
@property
|
|
def minimal_roi(self) -> Dict[str, float]:
|
|
"""
|
|
基于超参动态生成 ROI 梯度
|
|
"""
|
|
if self._roi_override is not None:
|
|
return self._roi_override
|
|
t1 = int(self.roi_t1.value)
|
|
t2 = int(self.roi_t2.value)
|
|
times = sorted([t1, t2])
|
|
p1 = float(self.roi_p1.value)
|
|
p2 = float(self.roi_p2.value)
|
|
profits = sorted([p1, p2], reverse=True)
|
|
return {
|
|
"0": profits[0],
|
|
str(times[0]): profits[1],
|
|
str(times[1]): 0.0,
|
|
}
|
|
|
|
@minimal_roi.setter
|
|
def minimal_roi(self, value: Dict[str, float]) -> None:
|
|
# 允许框架在解析时覆盖 ROI 设置
|
|
self._roi_override = value
|
|
|
|
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
"""
|
|
计算技术指标（极简版）
|
|
只计算双均线
|
|
"""
|
|
res = self.get_ticker_indicator() * int(self.time.value)
|
|
dataframe_3 = resample_to_interval(dataframe, res)
|
|
|
|
# 只计算双均线
|
|
dataframe_3['fast_ma'] = ta.SMA(dataframe_3['close'], timeperiod=int(self.fast_ma.value))
|
|
dataframe_3['slow_ma'] = ta.SMA(dataframe_3['close'], timeperiod=int(self.slow_ma.value))
|
|
|
|
# 计算金叉和死叉
|
|
dataframe_3['fast_ma_cross_slow_ma'] = (dataframe_3['fast_ma'] > dataframe_3['slow_ma']) & (dataframe_3['fast_ma'].shift(1) <= dataframe_3['slow_ma'].shift(1))
|
|
dataframe_3['fast_ma_cross_slow_ma_down'] = (dataframe_3['fast_ma'] < dataframe_3['slow_ma']) & (dataframe_3['fast_ma'].shift(1) >= dataframe_3['slow_ma'].shift(1))
|
|
|
|
dataframe = resampled_merge(dataframe, dataframe_3)
|
|
return dataframe
|
|
|
|
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
"""
|
|
基于TA指标，填充进场趋势列（极简版）
|
|
只使用双均线交叉
|
|
"""
|
|
res = self.get_ticker_indicator() * int(self.time.value)
|
|
def _pick(df: DataFrame, name: str) -> str:
|
|
col = f"resample_{res}_{name}"
|
|
if col in df.columns:
|
|
return col
|
|
col2 = f"resample_{float(res)}_{name}"
|
|
if col2 in df.columns:
|
|
return col2
|
|
cand = [c for c in df.columns if c.endswith(f"_{name}")]
|
|
return cand[0] if len(cand) else col
|
|
|
|
fast_ma_cross_slow_ma_str = _pick(dataframe, 'fast_ma_cross_slow_ma')
|
|
fast_ma_cross_slow_ma_down_str = _pick(dataframe, 'fast_ma_cross_slow_ma_down')
|
|
|
|
# 检测多头信号：快线上穿慢线
|
|
dataframe.loc[
|
|
(
|
|
(dataframe[fast_ma_cross_slow_ma_str] == True) &
|
|
(pd.notna(dataframe[fast_ma_cross_slow_ma_str]))
|
|
),
|
|
['enter_long', 'enter_tag']] = (1, 'long_signal_simple')
|
|
|
|
# 检测空头信号：快线下穿慢线
|
|
dataframe.loc[
|
|
(
|
|
(dataframe[fast_ma_cross_slow_ma_down_str] == True) &
|
|
(pd.notna(dataframe[fast_ma_cross_slow_ma_down_str]))
|
|
),
|
|
['enter_short', 'enter_tag']] = (1, 'short_signal_simple')
|
|
|
|
return dataframe
|
|
|
|
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
|
|
"""
|
|
基于TA指标，填充出场趋势列（极简版）
|
|
反向交叉出场
|
|
"""
|
|
res = self.get_ticker_indicator() * int(self.time.value)
|
|
def _pick(df: DataFrame, name: str) -> str:
|
|
col = f"resample_{res}_{name}"
|
|
if col in df.columns:
|
|
return col
|
|
col2 = f"resample_{float(res)}_{name}"
|
|
if col2 in df.columns:
|
|
return col2
|
|
cand = [c for c in df.columns if c.endswith(f"_{name}")]
|
|
return cand[0] if len(cand) else col
|
|
|
|
fast_ma_cross_slow_ma_str = _pick(dataframe, 'fast_ma_cross_slow_ma')
|
|
fast_ma_cross_slow_ma_down_str = _pick(dataframe, 'fast_ma_cross_slow_ma_down')
|
|
|
|
# 做多出场：出现死叉
|
|
dataframe.loc[
|
|
(
|
|
(dataframe[fast_ma_cross_slow_ma_down_str] == True) &
|
|
(pd.notna(dataframe[fast_ma_cross_slow_ma_down_str]))
|
|
),
|
|
['exit_long', 'exit_tag']] = (1, 'long_exit_simple')
|
|
|
|
# 做空出场：出现金叉
|
|
dataframe.loc[
|
|
(
|
|
(dataframe[fast_ma_cross_slow_ma_str] == True) &
|
|
(pd.notna(dataframe[fast_ma_cross_slow_ma_str]))
|
|
),
|
|
['exit_short', 'exit_tag']] = (1, 'short_exit_simple')
|
|
|
|
return dataframe
|
|
|
|
def leverage(self, pair: str, current_time: datetime, current_rate: float,
|
|
proposed_leverage: float, max_leverage: float, entry_tag: Optional[str], side: str,
|
|
**kwargs) -> float:
|
|
"""
|
|
杠杆设置
|
|
"""
|
|
return float(self.lev.value)
|
|
|
|
def get_ticker_indicator(self):
|
|
return int(self.timeframe[:-1])
|
|
|
|
|
|
# 简单的测试函数
|
|
def test_strategy():
|
|
"""
|
|
测试策略基本功能
|
|
"""
|
|
try:
|
|
# 创建策略实例
|
|
strategy = PatternTrader()
|
|
|
|
# 检查基本属性
|
|
print("✅ 策略实例化成功")
|
|
print(f"策略名称: {strategy.__class__.__name__}")
|
|
print(f"接口版本: {strategy.INTERFACE_VERSION}")
|
|
print(f"支持做空: {strategy.can_short}")
|
|
print(f"默认止损: {strategy.stoploss}")
|
|
|
|
# 检查参数
|
|
print("\n✅ 策略参数检查:")
|
|
print(f"布林带长度: {strategy.bb_length.value}")
|
|
print(f"杠杆: {strategy.lev.value}")
|
|
print(f"仓位比例: {strategy.position_size_pct.value}")
|
|
|
|
print("\n🎉 策略测试通过！")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"❌ 策略测试失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return False
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_strategy() |