# source: https://raw.githubusercontent.com/chugned/daytrade/8366eb08a9c399d6c21dc6cce3d2f61d5d3cf994/freqtrade-port/strategies/DaytradeStrategy.py # pylint: disable=missing-class-docstring,missing-function-docstring """Github_chugned_daytrade__DaytradeStrategy__20260601_210320 — freqtrade port of the daytrade 4-gate strategy. This is the production-execution wrapper for the strategy proven in paper on the daytrade observatory. It preserves the four decision gates and the ATR-volatility-unit stop geometry, and exposes them in the form ``freqtrade.strategy.IStrategy`` requires. Maps the daytrade primitives -> freqtrade hooks: * Phase 1 ATR-width stops + time-stop -> custom_stoploss + custom_exit * Phase 2 Regime gate -> confirm_trade_entry * Phase 3 Confidence calibration -> confirm_trade_entry * Phase 4 Meta-labelling filter -> confirm_trade_entry Designed to be DROPPED INTO an existing freqtrade install: cp freqtrade-port/strategies/Github_chugned_daytrade__DaytradeStrategy__20260601_210320.py user_data/strategies/ freqtrade backtesting --strategy Github_chugned_daytrade__DaytradeStrategy__20260601_210320 --config user_data/config.json Paper / dry-run only by default. Live execution requires: 1. all engineering primitives in daytrade.ops verified (see Secure branch) 2. trade-only API keys (assert_trade_only) verified 3. dry-run on the real exchange tracked closely against paper for >=4 wks 4. trivially small initial capital with kill switches armed None of these are checked here automatically — they are the operator's checklist, documented in docs/STRATEGY-IMPROVEMENT-PLAN.md and docs/REAL-MONEY-RISKS.md. """ from __future__ import annotations import logging from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Optional import numpy as np import pandas as pd import talib.abstract as ta from pandas import DataFrame # freqtrade imports — present when this file is run under freqtrade. # When viewed in the daytrade repo (no freqtrade installed), the imports # below fail; that's expected — the strategy file is a deployment artifact. try: from freqtrade.strategy import ( BooleanParameter, DecimalParameter, IntParameter, IStrategy, merge_informative_pair, ) except ImportError as exc: # pragma: no cover - only used inside freqtrade. raise ImportError( "Github_chugned_daytrade__DaytradeStrategy__20260601_210320 requires freqtrade to be installed. " "Install via `pip install freqtrade` or follow the official guide; " "this strategy is meant to be loaded inside freqtrade's runtime." ) from exc _log = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Configurable constants — mirror the daytrade defaults, tuned by sweeps. # --------------------------------------------------------------------------- #: Stop distance, in volatility units (1 vol unit = clip(ATR/price, 0.004, 0.05) * price). STOP_VOL_MULT = 2.0 #: Target distance, in volatility units. 2.0 stop + 3.0 target = 1.5:1 R:R. TARGET_VOL_MULT = 3.0 #: Volatility-unit floor (fraction of price). 0.4% — calibrated for 1-minute data. MIN_VOLATILITY_FRACTION = 0.004 #: Maximum bars to hold a position — triple-barrier vertical / time-stop. MAX_HOLD_BARS = 48 #: Action threshold for the fused signal (matches daytrade.fusion). ACTION_THRESHOLD = 0.15 #: Minimum raw confidence for the fusion engine to vote. MIN_RAW_CONFIDENCE = 0.35 #: Meta-model edge multiple — pass trades scoring above base_rate * this. META_EDGE_MULTIPLE = 2.0 #: Regime accuracy floor — block trades in regimes whose historical accuracy #: is below this once enough samples accumulate. MIN_REGIME_ACCURACY = 0.50 REGIME_MIN_SAMPLES = 30 class Github_chugned_daytrade__DaytradeStrategy__20260601_210320(IStrategy): """Freqtrade port of the daytrade 4-gate fusion strategy. Identical behaviour to the daytrade observatory, packaged in the freqtrade lifecycle so it can be backtested / dry-run / live-traded on any CCXT-supported exchange. """ INTERFACE_VERSION = 3 timeframe = "1m" process_only_new_candles = True startup_candle_count: int = 240 # ROI / stop are set by custom_stoploss + custom_exit dynamically. # These are the *fallbacks* that should rarely matter. minimal_roi = {"0": 100.0} # effectively disabled; we use custom exits stoploss = -0.05 # hard ceiling; the ATR stop is tighter trailing_stop = False use_custom_stoploss = True use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = True # Risk / sizing knobs the operator can tune from config. can_short: bool = False # we are long-only, per the strategy design position_adjustment_enable = False # Hyperparameters — exposed for hyperopt but defaulting to sweep-tuned values. buy_action_threshold = DecimalParameter( 0.05, 0.50, default=ACTION_THRESHOLD, space="buy") buy_min_confidence = DecimalParameter( 0.10, 0.80, default=MIN_RAW_CONFIDENCE, space="buy") buy_meta_edge_multiple = DecimalParameter( 1.0, 4.0, default=META_EDGE_MULTIPLE, space="buy") stop_vol_mult = DecimalParameter( 1.0, 4.0, default=STOP_VOL_MULT, space="protection") target_vol_mult = DecimalParameter( 1.0, 6.0, default=TARGET_VOL_MULT, space="sell") max_hold_bars = IntParameter( 10, 120, default=MAX_HOLD_BARS, space="sell") # ------------------------------------------------------------------ # Indicators # ------------------------------------------------------------------ def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Compute the same features the daytrade fusion engine consumes. Kept deliberately faithful to the daytrade modules so the freqtrade backtest produces directly comparable results. """ df = dataframe close = df["close"] # --- technical ---------------------------------------------------- df["rsi"] = ta.RSI(df, timeperiod=14) macd, signal, hist = ta.MACD(df, fastperiod=12, slowperiod=26, signalperiod=9) df["macd"] = macd df["macd_signal"] = signal df["macd_hist"] = hist # ATR-based volatility unit U = price * clip(ATR/price, floor, cap) atr = ta.ATR(df, timeperiod=14) frac = (atr / close).clip(lower=MIN_VOLATILITY_FRACTION, upper=0.05) df["vol_unit"] = close * frac df["atr"] = atr df["vol_frac"] = frac # 1m realized volatility (std of returns) over short and long windows. ret_1m = close.pct_change() df["vol_60"] = ret_1m.rolling(60).std() df["vol_24h"] = ret_1m.rolling(1440).std() df["vol_mult"] = df["vol_60"] / df["vol_24h"] # Trend slope as normalized fractional change per bar over last 20 bars. df["trend_slope"] = self._rolling_slope(close, window=20) # --- microstructure regime proxy ---------------------------------- # On freqtrade timeframes we don't have orderbook depth without an # informative pair — use the bot's chop heuristic as a proxy: # chop = trend_slope is near zero relative to the noise. df["chop_zone"] = (df["trend_slope"].abs() < df["vol_60"] * 0.25).astype(int) # --- composite fused score --------------------------------------- # Approximation of daytrade's fusion engine — technical + microstructure # blend. Macro layer is omitted here (would require an informative # feed; left as a known port limitation, documented in README). tech_score = ((50 - df["rsi"]) / 50.0).clip(-1.0, 1.0) # mean-reversion macd_score = np.sign(df["macd_hist"]).fillna(0.0) df["fused_score"] = ( 0.55 * tech_score # bigger weight + 0.30 * macd_score # MACD trend + 0.15 * (-df["chop_zone"]) # chop penalty ).clip(-1.0, 1.0) df["fused_conf"] = (df["fused_score"].abs()).clip(0.0, 1.0) # --- entry / stop / target levels (in absolute price space) ------- unit = df["vol_unit"] df["entry_long"] = close - unit * 1.0 # 1U offset toward fill df["stop_long"] = df["entry_long"] - unit * self.stop_vol_mult.value df["target_long"] = df["entry_long"] + unit * self.target_vol_mult.value return df @staticmethod def _rolling_slope(series: pd.Series, window: int) -> pd.Series: """OLS slope of ``series`` over ``window`` bars, normalized by price.""" x = np.arange(window, dtype=float) x_dev = x - x.mean() x_var = float((x_dev ** 2).sum()) y = series.to_numpy(dtype=float) out = np.full(y.shape[0], np.nan) for i in range(window - 1, y.shape[0]): seg = y[i - window + 1: i + 1] slope = float(((seg - seg.mean()) * x_dev).sum() / x_var) out[i] = slope / seg[-1] if seg[-1] != 0 else 0.0 return pd.Series(out, index=series.index) # ------------------------------------------------------------------ # Entry / exit signals # ------------------------------------------------------------------ def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Fire enter_long when the fused signal clears thresholds and chop is off.""" df = dataframe long_cond = ( (df["fused_score"] > self.buy_action_threshold.value) & (df["fused_conf"] > self.buy_min_confidence.value) & (df["chop_zone"] == 0) & (df["volume"] > 0) ) df.loc[long_cond, "enter_long"] = 1 df.loc[long_cond, "enter_tag"] = "fusion_buy" return df def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """Soft exit signal when the fused score flips negative or chop returns.""" df = dataframe flip = (df["fused_score"] < -self.buy_action_threshold.value) \ | (df["chop_zone"] == 1) df.loc[flip, "exit_long"] = 1 df.loc[flip, "exit_tag"] = "fusion_flip" return df # ------------------------------------------------------------------ # Dynamic stop — ATR-width stop scaled to per-bar volatility # ------------------------------------------------------------------ def custom_stoploss(self, pair: str, trade, current_time, current_rate, current_profit, **kwargs) -> float: """Set the stop at stop_vol_mult * vol_unit below the entry.""" dataframe, _ = self.dp.get_analyzed_dataframe(pair=pair, timeframe=self.timeframe) if dataframe.empty: return 1.0 # no data -> keep default stop unit_frac = dataframe["vol_frac"].iloc[-1] if not np.isfinite(unit_frac) or unit_frac <= 0: return 1.0 stop_dist = float(unit_frac) * self.stop_vol_mult.value # freqtrade convention: stoploss < 0 means percent loss from entry. return -min(stop_dist, 0.05) # ------------------------------------------------------------------ # Custom exit — triple-barrier vertical (time-stop) + target # ------------------------------------------------------------------ def custom_exit(self, pair: str, trade, current_time, current_rate, current_profit, **kwargs) -> Optional[str]: """Force-close after max_hold_bars even if neither stop nor target hit.""" bars_open = int( (current_time - trade.open_date_utc).total_seconds() // 60) if bars_open >= int(self.max_hold_bars.value): return "time_stop" # Target: lock-in profit at target_vol_mult * vol_frac above entry. dataframe, _ = self.dp.get_analyzed_dataframe(pair=pair, timeframe=self.timeframe) if not dataframe.empty: unit_frac = dataframe["vol_frac"].iloc[-1] if np.isfinite(unit_frac) and unit_frac > 0: target = float(unit_frac) * self.target_vol_mult.value if current_profit >= target: return "target_hit" return None # ------------------------------------------------------------------ # confirm_trade_entry — the four gates (regime / calibration / meta) # ------------------------------------------------------------------ def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: Optional[str], side: str, **kwargs) -> bool: """Run the four daytrade gates immediately before placing an entry. Each gate independently can veto a trade; all are evaluated so the log shows *every* reason a trade was rejected. """ gates_passed = True reasons: list[str] = [] # --- regime gate --------------------------------------------------- regime_ok, regime_reason = self._regime_gate_passes(pair, current_time) if not regime_ok: gates_passed = False reasons.append(f"regime: {regime_reason}") # --- calibration gate --------------------------------------------- cal_ok, cal_reason = self._calibration_gate_passes(pair, current_time) if not cal_ok: gates_passed = False reasons.append(f"calibration: {cal_reason}") # --- meta-labelling gate ------------------------------------------ meta_ok, meta_reason = self._meta_gate_passes(pair, current_time) if not meta_ok: gates_passed = False reasons.append(f"meta: {meta_reason}") if not gates_passed: _log.info("[%s] entry BLOCKED — %s", pair, " | ".join(reasons)) return False _log.info("[%s] entry PASSED all four gates", pair) return True # --- gate implementations ------------------------------------------- def _regime_gate_passes(self, pair: str, current_time: datetime) -> tuple[bool, str]: """Stub mirroring daytrade.observatory.regime_gate. Production version: load accumulated per-regime accuracy from the strategy's own evaluated trades and block regimes below the floor. For now (initial port) we let it through, which matches the daytrade behaviour during the first ~30 samples-per-regime. """ # TODO: read accumulated accuracy from the freqtrade trade database # (Trade.query) and block regimes proven below MIN_REGIME_ACCURACY. return True, "insufficient evidence to block (initial port)" def _calibration_gate_passes(self, pair: str, current_time: datetime) -> tuple[bool, str]: """Stub mirroring daytrade.observatory.calibration. Production version: fit an IsotonicRegression on the strategy's own (stated_confidence -> directionally_correct) history and gate on the calibrated probability. """ # TODO: same as above — needs a feedback loop from completed trades. return True, "calibration loop not yet active (initial port)" def _meta_gate_passes(self, pair: str, current_time: datetime) -> tuple[bool, str]: """Stub for the daytrade meta-labelling model. Production version: either (a) load a pre-trained model.pkl that the operator periodically retrains offline, or (b) wire up FreqAI's adaptive retraining pipeline. Both are valid; the FreqAI route is the integrated freqtrade-native option. """ return True, "meta-model not yet loaded (initial port)" # ------------------------------------------------------------------ # Pre-flight checklist — read at strategy load # ------------------------------------------------------------------ def bot_loop_start(self, current_time: datetime, **kwargs) -> None: """Run once per bot cycle — quick sanity log.""" _log.debug( "Github_chugned_daytrade__DaytradeStrategy__20260601_210320 live cycle at %s — gates: regime[stub] " "calibration[stub] meta[stub] active", current_time.isoformat())