MacroRegimeStrategy

"""
MacroRegimeStrategy — basic EMA cross with macro risk-on gate.

Thesis: Crypto trend strategies fail because they trade against the macro tape.
If we only enter on EMA cross when the macro is risk-on (SPX above 50d SMA AND
DXY below 50d SMA), the same signal might pass the gate.

Hypothesis is testable in a session where most of the window was risk-on (SPX
+60%, DXY -8%) — if regime filtering still doesn't help, the issue isn't macro
context.

Drafted alongside FearGreedStrategy as the second "untested signal class"
experiment after v5+v4+grid failed.
"""
from pathlib import Path

import pandas as pd
import talib.abstract as ta
from pandas import DataFrame
from freqtrade.strategy import IStrategy

# Configurable local data path (relative to repo root). Point this at your own
# SPX/DXY feather dump (see scripts/fetch_macro.py).
MACRO_PATH = Path("user_data/data/macro_history.feather")

EMA_FAST = 20
EMA_SLOW = 50
MACRO_SMA = 50

_MACRO_CACHE: DataFrame | None = None


def _load_macro() -> DataFrame:
    global _MACRO_CACHE
    if _MACRO_CACHE is None:
        df = pd.read_feather(MACRO_PATH).rename(columns={"index": "date"})
        df["date"] = pd.to_datetime(df["date"], utc=True)
        df["spx_sma"] = df["SPX"].rolling(MACRO_SMA).mean()
        df["dxy_sma"] = df["DXY"].rolling(MACRO_SMA).mean()
        df["risk_on"] = (df["SPX"] > df["spx_sma"]) & (df["DXY"] < df["dxy_sma"])
        df["date_only"] = df["date"].dt.normalize()
        _MACRO_CACHE = df[["date_only", "SPX", "DXY", "risk_on"]]
    return _MACRO_CACHE


class MacroRegimeStrategy(IStrategy):
    INTERFACE_VERSION = 3
    timeframe = "1h"

    minimal_roi = {"0": 0.08, "24": 0.05, "48": 0.03, "96": 0.02}
    stoploss = -0.08

    trailing_stop = True
    trailing_stop_positive = 0.02
    trailing_stop_positive_offset = 0.05
    trailing_only_offset_is_reached = True

    process_only_new_candles = True
    use_exit_signal = True
    exit_profit_only = False

    startup_candle_count = 100

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe["ema_fast"] = ta.EMA(dataframe["close"], timeperiod=EMA_FAST)
        dataframe["ema_slow"] = ta.EMA(dataframe["close"], timeperiod=EMA_SLOW)
        dataframe["ema_cross_up"] = (
            (dataframe["ema_fast"] > dataframe["ema_slow"])
            & (dataframe["ema_fast"].shift(1) <= dataframe["ema_slow"].shift(1))
        )
        dataframe["ema_cross_down"] = (
            (dataframe["ema_fast"] < dataframe["ema_slow"])
            & (dataframe["ema_fast"].shift(1) >= dataframe["ema_slow"].shift(1))
        )

        macro = _load_macro()
        dataframe["date_only"] = dataframe["date"].dt.normalize()
        merged = dataframe.merge(macro, on="date_only", how="left")
        merged["risk_on"] = merged["risk_on"].ffill().fillna(False)
        merged = merged.drop(columns=["date_only"])
        return merged

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe.loc[
            dataframe["ema_cross_up"] & dataframe["risk_on"],
            "enter_long",
        ] = 1
        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe.loc[dataframe["ema_cross_down"], "exit_long"] = 1
        return dataframe