utils/helpers.py

import math
import numbers
import numpy as np
import pandas as pd
MINUTES_PER_CANDLE = {'1m': 1, '2m': 2, '5m': 5, '10m': 10, '15m': 15, '30m': 30, '1h': 60, '2h': 120, '4h': 240, '1d': 1440}
INDICATOR_WARMUP_CANDLES = 30
PREDICTION_HISTORY_CANDLES = 200


def forex_market_open(timestamp):
    """Standard FX week: Sunday 17:00 to Friday 17:00 New York (DST aware)."""
    stamp=pd.Timestamp(timestamp)
    if stamp.tzinfo is None:
        stamp=stamp.tz_localize('UTC')
    stamp=stamp.tz_convert('America/New_York')
    weekday=stamp.weekday()
    return not (weekday==5 or (weekday==4 and stamp.hour>=17) or (weekday==6 and stamp.hour<17))

def infer_step(index: pd.DatetimeIndex):
    if not isinstance(index, pd.DatetimeIndex) or index.isna().any() or index.duplicated().any() or not index.is_monotonic_increasing:
        raise ValueError('infer_step needs ordered unique datetime timestamps.')
    if len(index) < 2:
        return pd.Timedelta(minutes=1)
    diffs = index.to_series().diff().dropna()
    if diffs.empty:
        return pd.Timedelta(minutes=1)
    return diffs.median()

def future_index(last_timestamp, step, n):
    if isinstance(n, (bool, np.bool_)) or not isinstance(n, numbers.Integral) or n < 0 or pd.Timedelta(step) <= pd.Timedelta(0):
        raise ValueError('Future index requires a non-negative integer horizon and positive step.')
    return [last_timestamp + step * (i + 1) for i in range(n)]

def history_days_for_candles(timeframe: str, candles: int, minimum_days: int=1) -> int:
    minutes = MINUTES_PER_CANDLE.get(timeframe, MINUTES_PER_CANDLE['1h']) * max(int(candles), 0)
    return max(minimum_days, math.ceil(minutes / (24 * 60)))

def feature_warmup_days(timeframe: str, minimum_days: int=2) -> int:
    return history_days_for_candles(timeframe, INDICATOR_WARMUP_CANDLES, minimum_days=minimum_days)

def drop_forming_candle(df: pd.DataFrame, features_df: pd.DataFrame=None, timeframe: str=None, now=None):
    if df is None or len(df) == 0:
        return (df, features_df)
    if not isinstance(df.index, pd.DatetimeIndex) or df.index.isna().any():
        raise ValueError('Closed-candle filtering requires valid datetime timestamps.')
    if features_df is not None and len(features_df) != len(df):
        raise ValueError('Feature rows must align with candles before closed-candle filtering.')
    now = pd.Timestamp(now) if now is not None else pd.Timestamp.now(tz='UTC')
    if df.index.tz is None:
        if now.tzinfo is not None:
            now = now.tz_convert('UTC').tz_localize(None)
    elif now.tzinfo is None:
        now = now.tz_localize('UTC')
    step = pd.Timedelta(minutes=MINUTES_PER_CANDLE[timeframe]) if timeframe else infer_step(df.index)
    mask = df.index + step <= now
    df = df.loc[mask]
    if features_df is not None:
        features_df = features_df.loc[mask].reset_index(drop=True)
    return (df, features_df)

def price_direction(predicted: float, current: float) -> str:
    # Treat optimizer-scale numerical noise as flat. Without this guard a
    # persistence forecast can become DOWN solely because exp/log round-trip
    # error is a few ulps, which makes live consensus and accuracy tables look
    # more certain than the underlying prices justify.
    if not math.isfinite(float(predicted)) or not math.isfinite(float(current)):
        raise ValueError('price_direction needs finite numeric prices.')
    predicted, current = float(predicted), float(current)
    if math.isclose(float(predicted), float(current), rel_tol=1e-9, abs_tol=1e-12):
        return 'flat'
    if predicted > current:
        return 'up'
    if predicted < current:
        return 'down'
    return 'flat'