Spaces:
Running
Running
File size: 3,682 Bytes
9831ced 665c301 9831ced | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 | 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'
|