254 lines
9.9 KiB
Python
254 lines
9.9 KiB
Python
"""
|
|
Default Meta-Trend Strategy
|
|
|
|
This module implements the default trading strategy based on meta-trend analysis
|
|
using multiple Supertrend indicators. The strategy enters when trends align
|
|
and exits on trend reversal or stop loss.
|
|
|
|
The meta-trend is calculated by comparing three Supertrend indicators:
|
|
- Entry: When meta-trend changes from != 1 to == 1
|
|
- Exit: When meta-trend changes to -1 or stop loss is triggered
|
|
"""
|
|
|
|
import numpy as np
|
|
from typing import Tuple, Optional, List
|
|
|
|
from .base import StrategyBase, StrategySignal
|
|
|
|
|
|
class DefaultStrategy(StrategyBase):
|
|
"""
|
|
Default meta-trend strategy implementation.
|
|
|
|
This strategy uses multiple Supertrend indicators to determine market direction.
|
|
It generates entry signals when all three Supertrend indicators align in an
|
|
upward direction, and exit signals when they reverse or stop loss is triggered.
|
|
|
|
The strategy works best on 15-minute timeframes but can be configured for other timeframes.
|
|
|
|
Parameters:
|
|
stop_loss_pct (float): Stop loss percentage (default: 0.03)
|
|
timeframe (str): Preferred timeframe for analysis (default: "15min")
|
|
|
|
Example:
|
|
strategy = DefaultStrategy(weight=1.0, params={"stop_loss_pct": 0.05, "timeframe": "15min"})
|
|
"""
|
|
|
|
def __init__(self, weight: float = 1.0, params: Optional[dict] = None):
|
|
"""
|
|
Initialize the default strategy.
|
|
|
|
Args:
|
|
weight: Strategy weight for combination (default: 1.0)
|
|
params: Strategy parameters including stop_loss_pct and timeframe
|
|
"""
|
|
super().__init__("default", weight, params)
|
|
|
|
def get_timeframes(self) -> List[str]:
|
|
"""
|
|
Get the timeframes required by the default strategy.
|
|
|
|
The default strategy works on a single timeframe (typically 15min)
|
|
but also needs 1min data for precise stop-loss execution.
|
|
|
|
Returns:
|
|
List[str]: List containing primary timeframe and 1min for stop-loss
|
|
"""
|
|
primary_timeframe = self.params.get("timeframe", "15min")
|
|
|
|
# Always include 1min for stop-loss precision, avoid duplicates
|
|
timeframes = [primary_timeframe]
|
|
if primary_timeframe != "1min":
|
|
timeframes.append("1min")
|
|
|
|
return timeframes
|
|
|
|
def initialize(self, backtester) -> None:
|
|
"""
|
|
Initialize meta trend calculation using Supertrend indicators.
|
|
|
|
Calculates the meta-trend by comparing three Supertrend indicators.
|
|
When all three agree on direction, meta-trend follows that direction.
|
|
Otherwise, meta-trend is neutral (0).
|
|
|
|
Args:
|
|
backtester: Backtest instance with OHLCV data
|
|
"""
|
|
from cycles.Analysis.supertrend import Supertrends
|
|
|
|
# First, resample the original 1-minute data to required timeframes
|
|
self._resample_data(backtester.original_df)
|
|
|
|
# Get the primary timeframe data for strategy calculations
|
|
primary_timeframe = self.get_timeframes()[0]
|
|
strategy_data = self.get_data_for_timeframe(primary_timeframe)
|
|
|
|
# Calculate Supertrend indicators on the primary timeframe
|
|
supertrends = Supertrends(strategy_data, verbose=False)
|
|
supertrend_results_list = supertrends.calculate_supertrend_indicators()
|
|
|
|
# Extract trend arrays from each Supertrend
|
|
trends = [st['results']['trend'] for st in supertrend_results_list]
|
|
trends_arr = np.stack(trends, axis=1)
|
|
|
|
# Calculate meta-trend: all three must agree for direction signal
|
|
meta_trend = np.where(
|
|
(trends_arr[:,0] == trends_arr[:,1]) & (trends_arr[:,1] == trends_arr[:,2]),
|
|
trends_arr[:,0],
|
|
0 # Neutral when trends don't agree
|
|
)
|
|
|
|
# Store in backtester for access during trading
|
|
# Note: backtester.df should now be using our primary timeframe
|
|
backtester.strategies["meta_trend"] = meta_trend
|
|
backtester.strategies["stop_loss_pct"] = self.params.get("stop_loss_pct", 0.03)
|
|
backtester.strategies["primary_timeframe"] = primary_timeframe
|
|
|
|
self.initialized = True
|
|
|
|
def get_entry_signal(self, backtester, df_index: int) -> StrategySignal:
|
|
"""
|
|
Generate entry signal based on meta-trend direction change.
|
|
|
|
Entry occurs when meta-trend changes from != 1 to == 1, indicating
|
|
all Supertrend indicators now agree on upward direction.
|
|
|
|
Args:
|
|
backtester: Backtest instance with current state
|
|
df_index: Current index in the primary timeframe dataframe
|
|
|
|
Returns:
|
|
StrategySignal: Entry signal if trend aligns, hold signal otherwise
|
|
"""
|
|
if not self.initialized:
|
|
return StrategySignal("HOLD", 0.0)
|
|
|
|
if df_index < 2: # shifting one index to prevent lookahead bias
|
|
return StrategySignal("HOLD", 0.0)
|
|
|
|
# Check for meta-trend entry condition
|
|
prev_trend = backtester.strategies["meta_trend"][df_index - 2]
|
|
curr_trend = backtester.strategies["meta_trend"][df_index - 1]
|
|
|
|
if prev_trend != 1 and curr_trend == 1:
|
|
# Strong confidence when all indicators align for entry
|
|
return StrategySignal("ENTRY", confidence=1.0)
|
|
|
|
return StrategySignal("HOLD", confidence=0.0)
|
|
|
|
def get_exit_signal(self, backtester, df_index: int) -> StrategySignal:
|
|
"""
|
|
Generate exit signal based on meta-trend reversal or stop loss.
|
|
|
|
Exit occurs when:
|
|
1. Meta-trend changes to -1 (trend reversal)
|
|
2. Stop loss is triggered based on price movement
|
|
|
|
Args:
|
|
backtester: Backtest instance with current state
|
|
df_index: Current index in the primary timeframe dataframe
|
|
|
|
Returns:
|
|
StrategySignal: Exit signal with type and price, or hold signal
|
|
"""
|
|
if not self.initialized:
|
|
return StrategySignal("HOLD", 0.0)
|
|
|
|
if df_index < 1:
|
|
return StrategySignal("HOLD", 0.0)
|
|
|
|
# Check for meta-trend exit signal
|
|
prev_trend = backtester.strategies["meta_trend"][df_index - 1]
|
|
curr_trend = backtester.strategies["meta_trend"][df_index]
|
|
|
|
if prev_trend != 1 and curr_trend == -1:
|
|
return StrategySignal("EXIT", confidence=1.0,
|
|
metadata={"type": "META_TREND_EXIT_SIGNAL"})
|
|
|
|
# Check for stop loss using 1-minute data for precision
|
|
stop_loss_result, sell_price = self._check_stop_loss(backtester)
|
|
if stop_loss_result:
|
|
return StrategySignal("EXIT", confidence=1.0, price=sell_price,
|
|
metadata={"type": "STOP_LOSS"})
|
|
|
|
return StrategySignal("HOLD", confidence=0.0)
|
|
|
|
def get_confidence(self, backtester, df_index: int) -> float:
|
|
"""
|
|
Get strategy confidence based on meta-trend strength.
|
|
|
|
Higher confidence when meta-trend is strongly directional,
|
|
lower confidence during neutral periods.
|
|
|
|
Args:
|
|
backtester: Backtest instance with current state
|
|
df_index: Current index in the primary timeframe dataframe
|
|
|
|
Returns:
|
|
float: Confidence level (0.0 to 1.0)
|
|
"""
|
|
if not self.initialized or df_index >= len(backtester.strategies["meta_trend"]):
|
|
return 0.0
|
|
|
|
curr_trend = backtester.strategies["meta_trend"][df_index]
|
|
|
|
# High confidence for strong directional signals
|
|
if curr_trend == 1 or curr_trend == -1:
|
|
return 1.0
|
|
|
|
# Low confidence for neutral trend
|
|
return 0.3
|
|
|
|
def _check_stop_loss(self, backtester) -> Tuple[bool, Optional[float]]:
|
|
"""
|
|
Check if stop loss is triggered based on price movement.
|
|
|
|
Uses 1-minute data for precise stop loss checking regardless of
|
|
the primary timeframe used for strategy signals.
|
|
|
|
Args:
|
|
backtester: Backtest instance with current trade state
|
|
|
|
Returns:
|
|
Tuple[bool, Optional[float]]: (stop_loss_triggered, sell_price)
|
|
"""
|
|
# Calculate stop loss price
|
|
stop_price = backtester.entry_price * (1 - backtester.strategies["stop_loss_pct"])
|
|
|
|
# Use 1-minute data for precise stop loss checking
|
|
min1_data = self.get_data_for_timeframe("1min")
|
|
if min1_data is None:
|
|
# Fallback to original_df if 1min timeframe not available
|
|
min1_data = backtester.original_df if hasattr(backtester, 'original_df') else backtester.min1_df
|
|
|
|
min1_index = min1_data.index
|
|
|
|
# Find data range from entry to current time
|
|
start_candidates = min1_index[min1_index >= backtester.entry_time]
|
|
if len(start_candidates) == 0:
|
|
return False, None
|
|
|
|
backtester.current_trade_min1_start_idx = start_candidates[0]
|
|
end_candidates = min1_index[min1_index <= backtester.current_date]
|
|
|
|
if len(end_candidates) == 0:
|
|
return False, None
|
|
|
|
backtester.current_min1_end_idx = end_candidates[-1]
|
|
|
|
# Check if any candle in the range triggered stop loss
|
|
min1_slice = min1_data.loc[backtester.current_trade_min1_start_idx:backtester.current_min1_end_idx]
|
|
|
|
if (min1_slice['low'] <= stop_price).any():
|
|
# Find the first candle that triggered stop loss
|
|
stop_candle = min1_slice[min1_slice['low'] <= stop_price].iloc[0]
|
|
|
|
# Use open price if it gapped below stop, otherwise use stop price
|
|
if stop_candle['open'] < stop_price:
|
|
sell_price = stop_candle['open']
|
|
else:
|
|
sell_price = stop_price
|
|
|
|
return True, sell_price
|
|
|
|
return False, None |