# source: https://raw.githubusercontent.com/davidcagoh/algo-traders/d386c57ba7d0b4966a05012b487aaf34f4e6589d/freqtrade-experiment/hmm-slope-experiment/research/strategies/HmmSmaSlopeV2LongShort.py # directory_url: https://github.com/davidcagoh/algo-traders/blob/main/freqtrade-experiment/hmm-slope-experiment/research/strategies/ # User: davidcagoh # Repository: algo-traders # --------------------""" Github_davidcagoh_algo_traders__HmmSmaSlopeV2LongShort__20260926_061503 — V2 long entries plus HmmSmaSlopeV2Short's short entries in a single strategy. BACKTEST-ONLY DRAFT. Combines rather than chooses: enter_long fires on V2's original condition (bull_prob crossing up through BULL_THRESHOLD), enter_short fires on the complementary condition (bull_prob crossing DOWN through 1 - BULL_THRESHOLD, equivalent to bear_prob = 1 - bull_prob crossing up through BEAR_THRESHOLD). Only one HMM fit is needed since bull_prob and bear_prob are complementary by construction (bull_states = mean>0, bear_states = the complement). Sizing mirrors each side's slope logic: long size scales with positive slope_pct, short size scales with negative slope_pct magnitude. Untested. HmmSmaSlopeV2Short alone failed decisively in backtest (bear window 2025-10-15 to 2026-05-09, market -38.61%: strategy -13.12%, MDD 16.07%, well past the 5.5% kill threshold) even in a market that should have favored it. This variant exists to check whether combining with V2's proven long side dilutes or compounds that failure, not because the short signal itself has been rehabilitated. """ from __future__ import annotations from datetime import datetime from typing import Optional import numpy as np import pandas as pd import talib.abstract as ta from pandas import DataFrame from freqtrade.strategy import IStrategy try: from hmmlearn.hmm import GaussianHMM _HMM_AVAILABLE = True except ImportError: _HMM_AVAILABLE = False # HMM block (matches HmmRegime4Rolling / V2 / V3) RETURN_WINDOW = 24 N_COMPONENTS = 4 BULL_THRESHOLD = 0.65 BEAR_THRESHOLD = 0.65 # applied to bear_prob = 1 - bull_prob EXIT_THRESHOLD = 0.45 FIT_WINDOW = 1000 REFIT_EVERY = 168 # SMA-slope block (matches SmaRegime180 / V2 / V3) SMA_PERIOD = 180 SLOPE_LOOKBACK = 6 SLOPE_STRONG = 0.005 MIN_SIZE_FACTOR = 0.0 class Github_davidcagoh_algo_traders__HmmSmaSlopeV2LongShort__20260926_061503(IStrategy): INTERFACE_VERSION = 3 can_short = True timeframe = "4h" startup_candle_count = max(FIT_WINDOW + RETURN_WINDOW, SMA_PERIOD + SLOPE_LOOKBACK) minimal_roi = {"0": 100} stoploss = -0.10 trailing_stop = False process_only_new_candles = True use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: if not _HMM_AVAILABLE: raise ImportError( "hmmlearn is required for Github_davidcagoh_algo_traders__HmmSmaSlopeV2LongShort__20260926_061503. " "Activate the freqtrade venv and run: pip install hmmlearn" ) # ---- SMA slope (matches SmaRegime180) ---- dataframe["sma180"] = ta.SMA(dataframe, timeperiod=SMA_PERIOD) dataframe["sma180_slope"] = ( dataframe["sma180"] - dataframe["sma180"].shift(SLOPE_LOOKBACK) ) dataframe["slope_pct"] = dataframe["sma180_slope"] / dataframe["sma180"] # Long size: positive slope -> full size. Short size: negative slope # magnitude -> full size. custom_stake_amount picks the right one by side. dataframe["long_size_factor"] = ( (dataframe["slope_pct"] / SLOPE_STRONG).clip(lower=MIN_SIZE_FACTOR, upper=1.0) ) dataframe["short_size_factor"] = ( (-dataframe["slope_pct"] / SLOPE_STRONG).clip(lower=MIN_SIZE_FACTOR, upper=1.0) ) # ---- Rolling HMM (matches HmmRegime4Rolling) ---- log_return = np.log( dataframe["close"] / dataframe["close"].shift(RETURN_WINDOW) ) log_vol = np.log(dataframe["volume"].clip(lower=1e-9)) log_vol_z = (log_vol - log_vol.mean()) / max(log_vol.std(), 1e-9) dataframe["_log_return"] = log_return dataframe["_log_vol_z"] = log_vol_z valid_mask = dataframe[["_log_return", "_log_vol_z"]].notna().all(axis=1) dataframe["bull_prob"] = np.nan valid_idx = np.where(valid_mask.values)[0] if len(valid_idx) < FIT_WINDOW + REFIT_EVERY: return dataframe X_full = dataframe[["_log_return", "_log_vol_z"]].values first_refit = valid_idx[0] + FIT_WINDOW last_row = len(dataframe) bull_prob = np.full(last_row, np.nan) for r in range(first_refit, last_row, REFIT_EVERY): fit_start = r - FIT_WINDOW X_fit = X_full[fit_start:r] if np.isnan(X_fit).any(): continue try: model = GaussianHMM( n_components=N_COMPONENTS, covariance_type="full", n_iter=200, random_state=42, ) model.fit(X_fit) except Exception: continue bull_states = [i for i in range(N_COMPONENTS) if model.means_[i, 0] > 0] if not bull_states: bull_states = [int(np.argmax(model.means_[:, 0]))] seg_end = min(r + REFIT_EVERY, last_row) for t in range(r, seg_end): if np.isnan(X_full[t]).any(): continue X_score = X_full[fit_start:t + 1] try: post = model.predict_proba(X_score)[-1] except Exception: continue bull_prob[t] = post[bull_states].sum() dataframe["bull_prob"] = bull_prob # bear_prob is exactly complementary since bull/bear states partition # all N_COMPONENTS states — no second fit needed. dataframe["bear_prob"] = 1.0 - dataframe["bull_prob"] return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ (dataframe["bull_prob"] >= BULL_THRESHOLD) & (dataframe["bull_prob"].shift(1) < BULL_THRESHOLD), "enter_long", ] = 1 dataframe.loc[ (dataframe["bear_prob"] >= BEAR_THRESHOLD) & (dataframe["bear_prob"].shift(1) < BEAR_THRESHOLD), "enter_short", ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ (dataframe["bull_prob"] < EXIT_THRESHOLD) | (dataframe["sma180_slope"] <= 0), "exit_long", ] = 1 dataframe.loc[ (dataframe["bear_prob"] < EXIT_THRESHOLD) | (dataframe["sma180_slope"] >= 0), "exit_short", ] = 1 return dataframe def custom_stake_amount( self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: Optional[float], max_stake: float, leverage: float, entry_tag: Optional[str], side: str, **kwargs, ) -> float: df, _ = self.dp.get_analyzed_dataframe(pair=pair, timeframe=self.timeframe) if df is None or df.empty: return proposed_stake col = "long_size_factor" if side == "long" else "short_size_factor" size_factor = df[col].iloc[-1] if pd.isna(size_factor) or size_factor <= 0: return 0.0 return proposed_stake * float(size_factor)