!pip install yfinance pandas numpy matplotlib seaborn scipy --quiet

import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from scipy import stats
import warnings

warnings.filterwarnings('ignore')
%matplotlib inline
plt.rcParams['figure.figsize'] = (14, 5)
plt.rcParams['axes.spines.top']   = False
plt.rcParams['axes.spines.right'] = False
sns.set_palette('muted')
print('Imports ready.')

START_DATE      = '2017-01-01'
DXY_TICKER      = 'DX-Y.NYB'
BTC_TICKER      = 'BTC-USD'
MA_FAST         = 20    # days for fast moving average
MA_SLOW         = 100   # days for slow moving average
ROLLING_WINDOW  = 60    # days for rolling correlation
USE_SYNTHETIC   = False

KEY_LEVELS = [
    {'level': 114.8, 'label': '2022 Peak (BTC low)', 'color': 'red'},
    {'level': 103.8, 'label': 'Major resistance', 'color': 'orange'},
    {'level': 100.0, 'label': 'Psychological', 'color': 'gray'},
    {'level': 89.2,  'label': '2021 Low',  'color': 'green'},
]

def fetch_dxy_btc(dxy_ticker: str, btc_ticker: str, start: str) -> pd.DataFrame:
    """
    Fetch and align DXY and BTC daily close prices.

    Parameters
    ----------
    dxy_ticker : str
        Yahoo Finance ticker for DXY futures ('DX-Y.NYB').
    btc_ticker : str
        Yahoo Finance ticker for Bitcoin.
    start : str
        Start date in 'YYYY-MM-DD' format.

    Returns
    -------
    pd.DataFrame
        Columns: dxy, btc — daily close prices.

    Notes
    -----
    DXY futures trade weekdays only; forward-filled for weekends so daily
    correlation with 24/7 BTC can be computed without introducing look-ahead bias
    on the BTC side. Consider using 5-day returns (weekly) for cleaner analysis.
    """
    raw = yf.download([dxy_ticker, btc_ticker], start=start, progress=False, auto_adjust=True)
    df = raw['Close'].copy()
    df.columns = ['btc', 'dxy'] if btc_ticker < dxy_ticker else ['dxy', 'btc']
    # Reorder reliably
    rename_map = {dxy_ticker: 'dxy', btc_ticker: 'btc'}
    df = raw['Close'].rename(columns=rename_map)
    df.index = pd.to_datetime(df.index)
    df = df.ffill().dropna()
    print(f'Fetched {len(df)} days of DXY and BTC data.')
    print(f'DXY range: {df["dxy"].min():.1f} – {df["dxy"].max():.1f}')
    return df


def generate_synthetic_dxy_btc(start: str, n_days: int = 2500) -> pd.DataFrame:
    """
    Generate synthetic DXY and BTC prices with realistic inverse relationship.

    Parameters
    ----------
    start : str
        Start date in 'YYYY-MM-DD' format.
    n_days : int
        Number of trading days to generate.

    Returns
    -------
    pd.DataFrame
        Synthetic dxy and btc price series.
    """
    np.random.seed(99)
    dates = pd.date_range(start, periods=n_days, freq='B')

    dxy_rets  = 0.0001 + 0.004 * np.random.randn(n_days)
    # BTC tends to move inversely with DXY, but with much higher vol
    noise     = np.random.randn(n_days)
    btc_rets  = 0.0005 - 3.0 * dxy_rets + 0.030 * noise

    # Inject the 2022 DXY surge / BTC crash period
    peak_start, peak_end = 1100, 1300
    dxy_rets[peak_start:peak_end] += 0.003
    btc_rets[peak_start:peak_end] -= 0.012

    return pd.DataFrame({
        'dxy': 95 * np.exp(np.cumsum(dxy_rets)),
        'btc': 10000 * np.exp(np.cumsum(btc_rets)),
    }, index=dates)


if USE_SYNTHETIC:
    df = generate_synthetic_dxy_btc(START_DATE)
    print('Using synthetic data.')
else:
    try:
        df = fetch_dxy_btc(DXY_TICKER, BTC_TICKER, START_DATE)
    except Exception as e:
        print(f'Live fetch failed ({e}). Falling back to synthetic.')
        df = generate_synthetic_dxy_btc(START_DATE)

print(df.tail(3))

def compute_dxy_signals(
    df: pd.DataFrame,
    ma_fast: int,
    ma_slow: int,
    roll_window: int
) -> pd.DataFrame:
    """
    Compute DXY trend indicators and BTC-DXY rolling correlation.

    Parameters
    ----------
    df : pd.DataFrame
        Columns: dxy, btc.
    ma_fast : int
        Fast moving average period in days.
    ma_slow : int
        Slow moving average period in days.
    roll_window : int
        Rolling window for DXY-BTC correlation.

    Returns
    -------
    pd.DataFrame
        Extended DataFrame with:
        - dxy_ma_fast, dxy_ma_slow: moving averages of DXY
        - dxy_trend: +1 (uptrend), -1 (downtrend), 0 (neutral)
        - dxy_roc_20d: 20-day rate of change of DXY (%)
        - corr_rolling: rolling correlation of log-returns
        - btc_dxy_divergence: z-score of DXY-BTC spread

    Notes
    -----
    Rate-of-change (ROC) divergence identifies moments where DXY and BTC move
    in the same direction (anomalous) or where one lags the other significantly.
    These divergences often precede mean-reversion trades.
    """
    out = df.copy()
    out['dxy_ma_fast'] = out['dxy'].rolling(ma_fast).mean()
    out['dxy_ma_slow'] = out['dxy'].rolling(ma_slow).mean()

    out['dxy_trend'] = 0
    out.loc[out['dxy_ma_fast'] > out['dxy_ma_slow'], 'dxy_trend'] =  1  # DXY uptrend → bearish BTC
    out.loc[out['dxy_ma_fast'] < out['dxy_ma_slow'], 'dxy_trend'] = -1  # DXY downtrend → bullish BTC

    out['dxy_roc_20d'] = out['dxy'].pct_change(20) * 100
    out['btc_roc_20d'] = out['btc'].pct_change(20) * 100

    log_rets = np.log(out[['dxy', 'btc']] / out[['dxy', 'btc']].shift(1))
    out['corr_rolling'] = log_rets['dxy'].rolling(roll_window).corr(log_rets['btc'])

    # Divergence: normalized spread between DXY ROC and BTC ROC
    spread = out['dxy_roc_20d'] + out['btc_roc_20d']  # should be ~0 if inversely correlated
    spread_roll_mean = spread.rolling(90).mean()
    spread_roll_std  = spread.rolling(90).std()
    out['btc_dxy_divergence'] = (spread - spread_roll_mean) / (spread_roll_std + 1e-8)

    return out


signals = compute_dxy_signals(df, MA_FAST, MA_SLOW, ROLLING_WINDOW)
print(f'DXY current trend: {signals["dxy_trend"].iloc[-1]:+d}')
print(f'DXY 20d ROC: {signals["dxy_roc_20d"].iloc[-1]:.2f}%')
print(f'DXY-BTC rolling corr ({ROLLING_WINDOW}d): {signals["corr_rolling"].iloc[-1]:.3f}')

def backtest_dxy_strategy(
    signals: pd.DataFrame,
    initial_capital: float = 10_000.0
) -> pd.DataFrame:
    """
    Backtest a DXY-trend-based BTC position sizing strategy.

    Strategy logic:
    - DXY downtrend (fast MA < slow MA): full 100% BTC long
    - DXY neutral / ranging: 50% BTC long
    - DXY uptrend (fast MA > slow MA): 0% — stay in cash

    Parameters
    ----------
    signals : pd.DataFrame
        Output of compute_dxy_signals().
    initial_capital : float
        Starting capital in USD.

    Returns
    -------
    pd.DataFrame
        Backtest results with columns: btc_ret, position, strategy_ret,
        cumulative_btc, cumulative_strategy, drawdown_strategy.

    Notes
    -----
    This is a simplified backtest — no transaction costs, no slippage.
    The position change is applied on the NEXT day's open to avoid look-ahead bias
    (the signal is observed at close, position entered at next open).
    Real execution would face bid-ask spread and crypto exchange fees (~0.1%).
    """
    bt = signals[['btc', 'dxy_trend']].dropna().copy()
    bt['btc_ret'] = bt['btc'].pct_change()

    # Position sizing: DXY downtrend → 1.0, neutral → 0.5, uptrend → 0.0
    position_map = {-1: 1.0, 0: 0.5, 1: 0.0}
    bt['position'] = bt['dxy_trend'].map(position_map).shift(1)  # shift to avoid look-ahead

    bt['strategy_ret'] = bt['position'] * bt['btc_ret']
    bt['cumulative_btc']      = (1 + bt['btc_ret']).cumprod()
    bt['cumulative_strategy'] = (1 + bt['strategy_ret']).cumprod()

    rolling_max = bt['cumulative_strategy'].cummax()
    bt['drawdown_strategy'] = (bt['cumulative_strategy'] - rolling_max) / rolling_max * 100

    bt = bt.dropna()
    total_return = (bt['cumulative_strategy'].iloc[-1] - 1) * 100
    btc_return   = (bt['cumulative_btc'].iloc[-1] - 1) * 100
    max_dd       = bt['drawdown_strategy'].min()
    ann_ret      = (bt['cumulative_strategy'].iloc[-1] ** (252 / len(bt)) - 1) * 100

    print(f'Strategy total return : {total_return:.1f}%')
    print(f'Buy-and-hold BTC total: {btc_return:.1f}%')
    print(f'Annualized return      : {ann_ret:.1f}%')
    print(f'Max drawdown           : {max_dd:.1f}%')
    return bt


bt_results = backtest_dxy_strategy(signals)

def plot_dxy_btc_overview(
    signals: pd.DataFrame,
    bt_results: pd.DataFrame,
    key_levels: list
) -> None:
    """
    Four-panel chart: DXY with MAs, BTC, rolling correlation, and strategy equity.

    Parameters
    ----------
    signals : pd.DataFrame
        Output of compute_dxy_signals().
    bt_results : pd.DataFrame
        Output of backtest_dxy_strategy().
    key_levels : list of dict
        DXY price levels to annotate: [{'level': float, 'label': str, 'color': str}].
    """
    fig, axes = plt.subplots(4, 1, figsize=(15, 16), sharex=True)

    # Panel 1: DXY with moving averages
    axes[0].plot(signals.index, signals['dxy'],         color='navy',   linewidth=1.5, label='DXY')
    axes[0].plot(signals.index, signals['dxy_ma_fast'], color='orange', linewidth=1.0, linestyle='--', label=f'MA{MA_FAST}')
    axes[0].plot(signals.index, signals['dxy_ma_slow'], color='red',    linewidth=1.0, linestyle='-',  label=f'MA{MA_SLOW}')
    for kl in key_levels:
        axes[0].axhline(kl['level'], color=kl['color'], alpha=0.4, linewidth=0.8, linestyle=':')
        axes[0].text(signals.index[-1], kl['level'], f" {kl['label']}", fontsize=7, color=kl['color'], va='center')
    axes[0].set_ylabel('DXY Level')
    axes[0].set_title('U.S. Dollar Index (DXY) with Trend Moving Averages')
    axes[0].legend()

    # Panel 2: BTC log scale
    axes[1].plot(signals.index, signals['btc'], color='orange', linewidth=1.5)
    axes[1].set_yscale('log')
    axes[1].set_ylabel('BTC Price (log)')
    axes[1].set_title('BTC-USD Price')

    # Shade DXY uptrend periods red on BTC panel
    in_uptrend = False
    start_date = None
    for date, trend in signals['dxy_trend'].items():
        if trend == 1 and not in_uptrend:
            in_uptrend = True
            start_date = date
        elif trend != 1 and in_uptrend:
            axes[1].axvspan(start_date, date, alpha=0.15, color='red')
            in_uptrend = False

    # Panel 3: Rolling correlation
    axes[2].plot(signals.index, signals['corr_rolling'], color='steelblue', linewidth=1.5)
    axes[2].axhline(0, color='black', linewidth=0.8, linestyle='--')
    axes[2].fill_between(signals.index, signals['corr_rolling'], 0,
                          where=(signals['corr_rolling'] < 0), alpha=0.2, color='red', label='Inverse (expected)')
    axes[2].fill_between(signals.index, signals['corr_rolling'], 0,
                          where=(signals['corr_rolling'] >= 0), alpha=0.2, color='green', label='Positive (unusual)')
    axes[2].set_ylim(-1, 1)
    axes[2].set_ylabel('Correlation')
    axes[2].set_title(f'Rolling {ROLLING_WINDOW}d DXY-BTC Log-Return Correlation')
    axes[2].legend()

    # Panel 4: Strategy equity curves
    axes[3].plot(bt_results.index, bt_results['cumulative_btc'] * 100,      color='orange', linewidth=1.5, label='Buy & Hold BTC')
    axes[3].plot(bt_results.index, bt_results['cumulative_strategy'] * 100, color='green',  linewidth=1.5, label='DXY Strategy')
    axes[3].set_yscale('log')
    axes[3].set_ylabel('Portfolio Value (log, base=100)')
    axes[3].set_title('Strategy vs Buy-and-Hold Performance')
    axes[3].legend()

    plt.tight_layout()
    plt.show()


plot_dxy_btc_overview(signals, bt_results, KEY_LEVELS)

def export_dxy_analysis(
    signals: pd.DataFrame,
    bt_results: pd.DataFrame
) -> None:
    """
    Export DXY signals and backtest results to CSV.

    Parameters
    ----------
    signals : pd.DataFrame
        DXY trend signals and correlation data.
    bt_results : pd.DataFrame
        Backtest equity curve and positions.
    """
    signals.to_csv('dxy_btc_signals.csv')
    bt_results.to_csv('dxy_strategy_backtest.csv')
    print('Exported: dxy_btc_signals.csv')
    print('Exported: dxy_strategy_backtest.csv')


export_dxy_analysis(signals, bt_results)