feat: add USDCHF data consolidation feature and update related processing logic

This commit is contained in:
Luke Mosoti
2026-04-06 21:33:14 +03:00
parent 9e7c426e1c
commit 906a32c953
9 changed files with 284 additions and 120 deletions
+15 -1
View File
@@ -460,6 +460,19 @@ def cmd_guardian(args):
)
def cmd_consolidate(args):
"""Consolidate all USDCHF CSV exports in data/raw/ into USDCHF_master.csv."""
from src.data_consolidator import consolidate_usdchf
result = consolidate_usdchf()
if result.empty:
logger.error(
"Consolidation produced no data. Place USDCHF CSV files in data/raw/ "
"with 'USDCHF' in the filename (e.g. USDCHF_5m_data.csv)."
)
else:
print(f"\nUSDCHF master built: {len(result)} rows → data/processed/USDCHF_master.csv\n")
def cmd_listen(args):
"""Start the Telegram remote control listener + nightly report scheduler."""
import threading
@@ -503,7 +516,7 @@ def main():
parser.add_argument(
"--mode",
choices=["process", "optimize", "train", "compare", "export", "report",
"sync_validate", "live", "audit", "guardian", "listen"],
"sync_validate", "live", "audit", "guardian", "listen", "consolidate"],
required=True,
)
parser.add_argument("--trials", type=int, default=250)
@@ -541,6 +554,7 @@ def main():
"audit": cmd_audit,
"guardian": cmd_guardian,
"listen": cmd_listen,
"consolidate": cmd_consolidate,
}[args.mode](args)
+11 -7
View File
@@ -65,8 +65,9 @@ def apply_session_limits(
dates,
account_size: float = 15.0,
hmm_states: np.ndarray = None,
tf: str = "H1",
) -> np.ndarray:
"""Cap signals to the adaptive daily limit based on account size and HMM state.
"""Cap signals to the adaptive daily limit based on account size, HMM state and TF.
Counts both BUY (+1) and SELL (-1) signals toward the daily cap.
@@ -74,6 +75,9 @@ def apply_session_limits(
- Bull / Bear: 3 signals/day
- Chop: 2 signals/day
M5 small accounts get a higher cap (4/day) to match the 288-bar/day
frequency and allow the optimizer to generate enough OOS trades.
Uses the majority HMM state within each calendar day to decide the cap.
"""
arm = AdaptiveRiskManager(account_size)
@@ -88,9 +92,9 @@ def apply_session_limits(
if hmm_states is not None and not arm.is_small_account:
day_states = hmm_states[start:end].astype(int)
day_mode = int(np.bincount(day_states).argmax())
limits = arm.get_trade_limits(day_mode)
limits = arm.get_trade_limits(day_mode, tf=tf)
else:
limits = arm.get_trade_limits()
limits = arm.get_trade_limits(tf=tf)
max_daily = limits["max_daily_trades"]
seg = result[start:end]
@@ -181,11 +185,11 @@ def vectorized_backtest(
threshold=buy_th,
short_threshold=short_threshold, # None → symmetric default inside
)
signals = apply_session_limits(raw_signals, df.index, account_size, hmm_states)
signals = apply_session_limits(raw_signals, df.index, account_size, hmm_states, tf=tf)
# Determine pos_per_trade from the adaptive risk manager
arm = AdaptiveRiskManager(account_size)
base_limits = arm.get_trade_limits()
base_limits = arm.get_trade_limits(tf=tf)
pos_per_trade = base_limits["pos_per_trade"]
sizes = compute_position_sizes(signals, atr_norm, pos_per_trade=pos_per_trade)
@@ -201,7 +205,7 @@ def vectorized_backtest(
"account_size": account_size,
"broker": broker,
"tf": tf,
"session_max_trades": arm.get_trade_limits()["max_daily_trades"],
"session_max_trades": arm.get_trade_limits(tf=tf)["max_daily_trades"],
"pos_per_trade": pos_per_trade,
})
@@ -233,7 +237,7 @@ def vectorized_backtest(
broker,
account_size,
"small" if arm.is_small_account else "growth",
arm.get_trade_limits()["max_daily_trades"],
arm.get_trade_limits(tf=tf)["max_daily_trades"],
pos_per_trade,
BROKER_CONFIGS.get(broker, {}).get("spread_frac", 0)
+ BROKER_CONFIGS.get(broker, {}).get("commission_frac", 0),
+110
View File
@@ -0,0 +1,110 @@
"""USDCHF data consolidator.
Merges all USDCHF raw CSV exports from ``data/raw/`` into a single sorted
master file at ``data/processed/USDCHF_master.csv``. The master file is used
by the training pipeline as an intraday DXY proxy (USDCHF correlates ~0.85
with DXY for XAUUSD signals and is always available on Headway as a standard
Forex pair).
Usage:
python main.py --mode consolidate
# or directly:
python -c "from src.data_consolidator import consolidate_usdchf; consolidate_usdchf()"
"""
import pandas as pd
from pathlib import Path
from src.logger import setup_logger
logger = setup_logger(__name__)
RAW_DIR = Path("data/raw")
MASTER_PATH = Path("data/processed/USDCHF_master.csv")
def _read_usdchf_csv(f: Path) -> pd.DataFrame | None:
"""Read a single USDCHF CSV, auto-detecting delimiter and date format."""
try:
# Detect separator from the first line
header = f.read_text(encoding="utf-8", errors="replace").splitlines()[0]
sep = ";" if ";" in header else ","
# Try parsing as-is first; fall back to infer_datetime_format
for fmt in ["%m/%d/%Y", "%Y.%m.%d %H:%M", "%Y-%m-%d", None]:
try:
kwargs = dict(sep=sep, parse_dates=["Date"])
if fmt:
kwargs["date_format"] = fmt
df = pd.read_csv(f, **kwargs)
df.set_index("Date", inplace=True)
if pd.api.types.is_datetime64_any_dtype(df.index):
logger.info("Loaded %d rows from %s", len(df), f)
return df
except Exception:
continue
logger.warning("Could not parse dates in %s — skipping.", f)
return None
except Exception as exc:
logger.warning("Failed to read %s: %s — skipping.", f, exc)
return None
def consolidate_usdchf(
raw_dir: Path = RAW_DIR,
out_path: Path = MASTER_PATH,
) -> pd.DataFrame:
"""Consolidate all USDCHF CSV files from *raw_dir* into a single master.
Scans for files matching ``*USDCHF*.csv`` (case-insensitive) in *raw_dir*.
Supports both comma-delimited annual exports (``MM/DD/YYYY`` dates) and
semicolon-delimited MT5 exports (``YYYY.MM.DD HH:MM`` dates). The format
is auto-detected per file so mixed sources work together.
Duplicate timestamps are de-duplicated and the result is sorted
chronologically before saving.
Args:
raw_dir: Directory to scan for USDCHF CSV exports.
out_path: Destination for the consolidated master CSV.
Returns:
The consolidated DataFrame (also persisted to *out_path*).
"""
# Case-insensitive glob for Windows and Linux
files = sorted(
f for f in raw_dir.glob("*.csv")
if "usdchf" in f.name.lower()
)
if not files:
logger.warning(
"No USDCHF CSV files found in %s. "
"Export USDCHF data from MT5 History Center (any timeframe that "
"matches your trading TF) and place in data/raw/ with 'USDCHF' "
"in the filename (e.g. USDCHF_5m_data.csv).",
raw_dir,
)
return pd.DataFrame()
frames = []
for f in files:
df = _read_usdchf_csv(f)
if df is not None:
frames.append(df)
if not frames:
logger.error("All USDCHF source files failed to load — no master produced.")
return pd.DataFrame()
merged = pd.concat(frames)
merged = merged[~merged.index.duplicated(keep="last")]
merged.sort_index(inplace=True)
out_path.parent.mkdir(parents=True, exist_ok=True)
merged.to_csv(out_path)
logger.info(
"USDCHF master saved: %d rows -> %s (from %d source file(s))",
len(merged), out_path, len(files),
)
return merged
+9 -7
View File
@@ -13,9 +13,11 @@ ONNX_PATH = Path("models/xgb_model.onnx")
XGB_PKL_PATH = Path("models/xgb_model.pkl")
ENSEMBLE_PKL_PATH = Path("models/xgb_ensemble.pkl")
# Base features always present; dxy_log_return is added when DXY data is available.
FEATURE_COLS = ["hmm_state", "rsi_slope", "atr_normalized", "prev_log_return"]
DXY_FEATURE = "dxy_log_return"
# Base features always present; usdchf_log_return is added when the USDCHF
# master CSV is available (run python main.py --mode consolidate first).
FEATURE_COLS = ["hmm_state", "rsi_slope", "atr_normalized", "prev_log_return"]
USDCHF_FEATURE = "usdchf_log_return"
DXY_FEATURE = USDCHF_FEATURE # legacy alias — kept so old imports don't crash
# Volatility bucket labels (ATR tertiles: low / med / high)
VOL_BUCKETS = ["low", "med", "high"]
@@ -24,12 +26,12 @@ VOL_BUCKETS = ["low", "med", "high"]
def get_feature_cols(df: pd.DataFrame) -> list[str]:
"""Return the feature column list for this DataFrame.
Includes ``dxy_log_return`` only when the column is present and has enough
non-null values to be useful (>50% coverage).
Includes ``usdchf_log_return`` only when the column is present and has
enough non-null values to be useful (>50% coverage).
"""
cols = list(FEATURE_COLS)
if DXY_FEATURE in df.columns and df[DXY_FEATURE].notna().mean() > 0.5:
cols.append(DXY_FEATURE)
if USDCHF_FEATURE in df.columns and df[USDCHF_FEATURE].notna().mean() > 0.5:
cols.append(USDCHF_FEATURE)
return cols
+34 -35
View File
@@ -22,8 +22,8 @@ from src.notifier import send_telegram_msg
from src.logger import setup_logger
from src.processor import (
TF_CONFIG,
DXY_RAW_PATH,
load_dxy_data,
USDCHF_MASTER_PATH,
load_usdchf_data,
compute_log_returns,
kalman_smooth,
compute_volatility,
@@ -34,7 +34,6 @@ from src.engine_hmm import load_model as load_hmm, predict_states
from src.engine_xgb import (
load_xgb_ensemble, get_predictions_ensemble, assign_vol_bucket, FEATURE_COLS,
)
from src.mt5_sync import DXY_SYMBOL_ALIASES, _find_dxy_symbol
from src.risk_manager import AdaptiveRiskManager, CENT_MULTIPLIER
logger = setup_logger(__name__)
@@ -75,24 +74,24 @@ TF_TP_CONFIG = {
# Lazy MT5 timeframe map
# ── DXY fallback cache ────────────────────────────────────────────────────────
# When the broker doesn't carry DXY as a live symbol, we carry forward the
# last known daily log return from DXY_data.csv (much closer to reality than 0.0).
_DXY_FALLBACK_CACHE: float | None = None
# ── USDCHF fallback cache ─────────────────────────────────────────────────────
# When the live USDCHF bar fetch fails (e.g. symbol not subscribed), carry the
# last known return from the master CSV — far better than assuming 0.0.
_USDCHF_FALLBACK_CACHE: float | None = None
def _get_dxy_fallback() -> float:
"""Return the last known DXY log return from DXY_data.csv (cached after first load)."""
global _DXY_FALLBACK_CACHE
if _DXY_FALLBACK_CACHE is None:
def _get_usdchf_fallback() -> float:
"""Return the last known USDCHF log return from the master CSV (cached)."""
global _USDCHF_FALLBACK_CACHE
if _USDCHF_FALLBACK_CACHE is None:
try:
dxy_df = load_dxy_data(DXY_RAW_PATH)
if dxy_df is not None:
_DXY_FALLBACK_CACHE = float(dxy_df["dxy_log_return"].dropna().iloc[-1])
usdchf_df = load_usdchf_data(USDCHF_MASTER_PATH)
if usdchf_df is not None:
_USDCHF_FALLBACK_CACHE = float(usdchf_df["usdchf_log_return"].dropna().iloc[-1])
else:
_DXY_FALLBACK_CACHE = 0.0
_USDCHF_FALLBACK_CACHE = 0.0
except Exception:
_DXY_FALLBACK_CACHE = 0.0
return _DXY_FALLBACK_CACHE
_USDCHF_FALLBACK_CACHE = 0.0
return _USDCHF_FALLBACK_CACHE
_MT5_TF_MAP: dict | None = None
@@ -221,19 +220,19 @@ def check_margin(symbol: str, lot: float, order_type: int, price: float) -> bool
# Signal derivation
# ─────────────────────────────────────────────────────────────────────────────
def _fetch_dxy_log_return(tf_mt5: int, mt5) -> float | None:
"""Fetch the most recently completed DXY bar log return for live inference.
def _fetch_usdchf_log_return(tf_mt5: int, mt5) -> float | None:
"""Fetch the most recently completed USDCHF bar log return for live inference.
Returns ``None`` if no DXY symbol is found on the broker or data is
unavailable — callers should substitute ``0.0`` in that case.
USDCHF is always available on Headway as a standard Forex pair — unlike
DXY/USDX which many brokers don't carry. Returns ``None`` only if the
data call fails (symbol not subscribed, no recent bars, etc.).
"""
dxy_sym = _find_dxy_symbol(mt5)
if dxy_sym is None:
return None
rates = mt5.copy_rates_from_pos(dxy_sym, tf_mt5, 1, 3) # 3 completed bars
rates = mt5.copy_rates_from_pos("USDCHF", tf_mt5, 1, 3) # 3 completed bars
if rates is None or len(rates) < 2:
return None
closes = [r["close"] for r in rates]
if closes[-2] <= 0:
return None
return float(np.log(closes[-1] / closes[-2]))
@@ -392,8 +391,8 @@ def compute_live_features(
atr_price (float): ATR in price terms for SL calculation.
``feature_cols`` is loaded from the ensemble metadata so the live vector
always matches what the models were trained on. If ``dxy_log_return`` is
required and unavailable from the broker, it is filled with 0.0.
always matches what the models were trained on. If ``usdchf_log_return`` is
required and the fetch fails, the last known value from the master CSV is used.
"""
if feature_cols is None:
feature_cols = FEATURE_COLS
@@ -413,18 +412,18 @@ def compute_live_features(
"prev_log_return": prev_log_return,
}
if "dxy_log_return" in feature_cols:
if "usdchf_log_return" in feature_cols:
if mt5 is not None:
dxy_ret = _fetch_dxy_log_return(tf_mt5, mt5)
usdchf_ret = _fetch_usdchf_log_return(tf_mt5, mt5)
else:
dxy_ret = None
if dxy_ret is None:
dxy_ret = _get_dxy_fallback()
usdchf_ret = None
if usdchf_ret is None:
usdchf_ret = _get_usdchf_fallback()
logger.warning(
"DXY return unavailable — using last known value %.6f as fallback for dxy_log_return.",
dxy_ret,
"USDCHF return unavailable — using last known value %.6f as fallback.",
usdchf_ret,
)
feature_dict["dxy_log_return"] = dxy_ret
feature_dict["usdchf_log_return"] = usdchf_ret
features_df = pd.DataFrame([feature_dict])[feature_cols]
atr_price = atr_normalized * float(df["Close"].iloc[-1])
+20 -8
View File
@@ -64,12 +64,18 @@ def _get_tier(balance: float) -> str:
def _score_result(result: dict, tier: str, broker: str) -> float:
cfg = TIER_CONFIGS[tier]
if result["max_drawdown"] > cfg["dd_limit"]:
return -10.0
if broker == "headway_cent":
return result["sharpe_ratio"] - result["max_drawdown"] * cfg["dd_penalty"]
return result["sharpe_ratio"]
cfg = TIER_CONFIGS[tier]
dd = result["max_drawdown"]
base = (result["sharpe_ratio"] - dd * cfg["dd_penalty"]
if broker == "headway_cent"
else result["sharpe_ratio"])
if dd > cfg["dd_limit"]:
# Sliding penalty: -2.0 per 1% of DD over the limit.
# Replaces the hard -10.0 so trials with, e.g., 16% DD on a 15% limit
# can still score positively instead of being discarded outright.
overshoot = (dd - cfg["dd_limit"]) * 200 # 0.01 excess → -2.0
return base - overshoot
return base
def make_objective(balance: float = 15.0, broker: str = "standard", tf: str = "H1"):
@@ -104,14 +110,20 @@ def make_objective(balance: float = 15.0, broker: str = "standard", tf: str = "H
if short_threshold >= prob_threshold:
return -10.0
max_depth = trial.suggest_int("max_depth", 3, 5)
# M5 uses shallower trees (2-3) to prevent IS memorisation across the
# large bar count; heavier L1 reg (1-20) to sparsify feature weights.
if tf.upper() == "M5":
max_depth = trial.suggest_int("max_depth", 2, 3)
reg_alpha = trial.suggest_float("reg_alpha", 1.0, 20.0, log=True)
else:
max_depth = trial.suggest_int("max_depth", 3, 5)
reg_alpha = trial.suggest_float("reg_alpha", 0.01, 10.0, log=True)
learning_rate = trial.suggest_float("learning_rate", 0.01, 0.3, log=True)
n_estimators = trial.suggest_int("n_estimators", 100, 500, step=50)
subsample = trial.suggest_float("subsample", 0.6, 1.0)
colsample_bytree = trial.suggest_float("colsample_bytree", 0.5, 1.0)
min_child_weight = trial.suggest_int("min_child_weight", 1, 15)
gamma = trial.suggest_float("gamma", 0.01, 5.0, log=True)
reg_alpha = trial.suggest_float("reg_alpha", 0.01, 10.0, log=True)
try:
df = process_pipeline(
+71 -53
View File
@@ -10,9 +10,13 @@ logger = setup_logger(__name__)
# in raw observations). M15 is ~4× noisier than H1; M5 uses a low obs_cov
# (0.05) to keep the filter very responsive for fast scalping regime detection.
# n_states_default: M5 uses 4 states to capture micro-noise; M15/H1 use 3.
# Single DXY file — DXY is a daily index; one complete file is shared across all TFs.
# Export from MT5 History Center (XAUUSD → H1, then switch to DXY/USDX) and save here.
DXY_RAW_PATH = Path("data/raw/DXY_data.csv")
# Single shared USDCHF master file — intraday proxy for DXY (correlates ~0.85
# with USD Index for XAUUSD signals and is always available on Headway).
# Run python main.py --mode consolidate to build this from MT5 exports.
USDCHF_MASTER_PATH = Path("data/processed/USDCHF_master.csv")
# Legacy alias kept for any callers that still reference the old name directly.
DXY_RAW_PATH = USDCHF_MASTER_PATH
TF_CONFIG = {
"H1": {
@@ -58,50 +62,66 @@ def load_raw_data(path: Path = RAW_PATH) -> pd.DataFrame:
return df
def load_dxy_data(path: Path) -> pd.DataFrame | None:
"""Load a DXY CSV and return a single-column DataFrame with ``dxy_log_return``.
def load_usdchf_data(path: Path) -> pd.DataFrame | None:
"""Load the USDCHF master CSV and return a ``usdchf_log_return`` column.
Returns ``None`` if the file does not exist — callers treat this as
"DXY feature not available for this training run".
"USDCHF feature not available for this training run".
Expected format: same semicolon-delimited MT5 export as XAUUSD, with a
``Close`` column. File names follow the pattern ``DXY_{tf}_data.csv``
(e.g. ``data/raw/DXY_1h_data.csv``).
Run ``python main.py --mode consolidate`` to build the master from MT5
exports before training.
"""
if not path.exists():
return None
# DXY_data.csv is saved as YYYY-MM-DD (daily). Use infer to be format-agnostic.
df = pd.read_csv(path, sep=";", parse_dates=["Date"])
df.set_index("Date", inplace=True)
df = pd.read_csv(path, index_col="Date", parse_dates=True)
df.sort_index(inplace=True)
df["dxy_log_return"] = np.log(df["Close"] / df["Close"].shift(1))
logger.info("Loaded DXY data: %d rows from %s", len(df), path)
return df[["dxy_log_return"]]
df["usdchf_log_return"] = np.log(df["Close"] / df["Close"].shift(1))
logger.info("Loaded USDCHF master: %d rows from %s", len(df), path)
return df[["usdchf_log_return"]]
def map_dxy_to_bars(df_index: pd.DatetimeIndex, dxy_df: pd.DataFrame) -> pd.Series:
"""Map daily DXY log return onto an intraday bar index.
def map_usdchf_to_bars(df_index: pd.DatetimeIndex, usdchf_df: pd.DataFrame) -> pd.Series:
"""Align USDCHF log returns onto XAUUSD bar timestamps.
Handles two gap cases:
- Internal gaps (weekends / holidays inside the DXY date range) — filled via ffill.
- Future dates beyond the DXY file's end (e.g. live sync data newer than the
last DXY export) — the last known daily return is carried forward.
Also handles tz-aware indices (e.g. UTC from MT5 sync) by stripping timezone
before comparison with tz-naive DXY dates.
Handles both daily USDCHF data (e.g. annual CSV exports) and intraday data
(e.g. direct M5 MT5 export):
- Daily: normalise each bar's timestamp to midnight and ffill by date.
- Intraday: reindex directly onto bar timestamps with ffill.
"""
series = dxy_df["dxy_log_return"].copy()
# Strip timezone so tz-aware sync data can be compared against tz-naive DXY dates
normalized = df_index.tz_localize(None) if df_index.tz is not None else df_index
normalized = normalized.normalize()
max_date = normalized.max()
if max_date > series.index.max():
extension = pd.date_range(
series.index.max() + pd.Timedelta(days=1), max_date, freq="D"
)
series = pd.concat([series, pd.Series(series.iloc[-1], index=extension)])
series = series.ffill()
return normalized.map(series)
series = usdchf_df["usdchf_log_return"].copy()
# Strip timezone for tz-aware MT5 sync indices
idx = df_index.tz_localize(None) if df_index.tz is not None else df_index
# Detect daily vs intraday by checking whether all times are midnight
is_daily = (series.index == series.index.normalize()).all()
if is_daily:
# Daily data — map each intraday bar onto its calendar date
# Extend the series forward to cover any bars beyond the last date
normalized = idx.normalize()
max_date = normalized.max()
if max_date > series.index.max():
extension = pd.date_range(
series.index.max() + pd.Timedelta(days=1), max_date, freq="D"
)
series = pd.concat([series, pd.Series(series.iloc[-1], index=extension)])
series = series.ffill()
return normalized.map(series)
else:
# Intraday data — forward-fill onto bar timestamps directly
return series.reindex(idx, method="ffill")
# Legacy aliases — kept so old imports don't crash if any caller still uses them.
def load_dxy_data(path: Path) -> pd.DataFrame | None: # noqa: D103
return load_usdchf_data(path)
def map_dxy_to_bars(df_index: pd.DatetimeIndex, dxy_df: pd.DataFrame) -> pd.Series: # noqa: D103
# Adapt old dxy_log_return column name if present (backwards compat)
if "dxy_log_return" in dxy_df.columns and "usdchf_log_return" not in dxy_df.columns:
dxy_df = dxy_df.rename(columns={"dxy_log_return": "usdchf_log_return"})
return map_usdchf_to_bars(df_index, dxy_df)
def filter_data(df: pd.DataFrame, years: int = 10) -> pd.DataFrame:
@@ -192,23 +212,21 @@ def process_pipeline(
df["rsi_slope"] = df["rsi"].diff()
df["atr_normalized"] = compute_atr(df)
# ── Optional DXY cross-asset feature ────────────────────────────────────
# DXY is a daily index shared across all TFs — one file covers H1/M15/M5.
dxy_path = DXY_RAW_PATH
if dxy_path:
dxy_df = load_dxy_data(dxy_path)
if dxy_df is not None:
# DXY is daily — normalize each intraday bar's timestamp to midnight
# so it matches the DXY date index before mapping.
df["dxy_log_return"] = map_dxy_to_bars(df.index, dxy_df)
n_dxy = df["dxy_log_return"].notna().sum()
logger.info("DXY merged: %d non-null dxy_log_return values.", n_dxy)
else:
logger.info(
"DXY file not found at %s — pipeline running without cross-asset feature. "
"Export DXY from MT5 History Center to enable it.",
dxy_path,
)
# ── Optional USDCHF cross-asset feature ─────────────────────────────────
# USDCHF is an intraday DXY proxy — correlates ~0.85 with USD Index and
# is always available on Headway as a standard Forex pair.
# Build the master with: python main.py --mode consolidate
usdchf_df = load_usdchf_data(USDCHF_MASTER_PATH)
if usdchf_df is not None:
df["usdchf_log_return"] = map_usdchf_to_bars(df.index, usdchf_df)
n_usdchf = df["usdchf_log_return"].notna().sum()
logger.info("USDCHF merged: %d non-null usdchf_log_return values.", n_usdchf)
else:
logger.info(
"USDCHF master not found at %s — pipeline running with 4 base features. "
"Run python main.py --mode consolidate to enable the 5th feature.",
USDCHF_MASTER_PATH,
)
df.dropna(inplace=True)
logger.info(
+7 -2
View File
@@ -105,15 +105,20 @@ class AdaptiveRiskManager:
def is_small_account(self) -> bool:
return self.balance <= SMALL_ACCOUNT_THRESHOLD
def get_trade_limits(self, market_state: int = None) -> dict:
def get_trade_limits(self, market_state: int = None, tf: str = None) -> dict:
"""Return trade limits for the current balance and optional HMM state.
Args:
market_state: Current HMM state index. If None, uses the most
permissive limit for the account tier.
tf: Timeframe string (e.g. ``"M5"``). M5 small accounts get a
higher daily cap (4) to match the higher bar frequency.
"""
if self.is_small_account:
return {"max_daily_trades": 2, "pos_per_trade": 1, "total_daily_pos": 2}
# M5 generates ~288 bars/day — allow more signals so the optimizer
# can find frequent enough OOS trades to clear MIN_OOS_TRADES=300.
max_daily = 4 if (tf and tf.upper() == "M5") else 2
return {"max_daily_trades": max_daily, "pos_per_trade": 1, "total_daily_pos": max_daily}
# Growth account: market-state-dependent
in_chop = (market_state == self.CHOP_STATE) if market_state is not None else False
+7 -7
View File
@@ -21,9 +21,9 @@ from pathlib import Path
from src.logger import setup_logger
from src.processor import (
TF_CONFIG,
DXY_RAW_PATH,
load_dxy_data,
map_dxy_to_bars,
USDCHF_MASTER_PATH,
load_usdchf_data,
map_usdchf_to_bars,
compute_log_returns,
kalman_smooth,
compute_volatility,
@@ -71,10 +71,10 @@ def _apply_features(
df["rsi_slope"] = df["rsi"].diff()
df["atr_normalized"] = compute_atr(df)
# Mirror process_pipeline: add DXY if the shared file exists
dxy_df = load_dxy_data(DXY_RAW_PATH)
if dxy_df is not None:
df["dxy_log_return"] = map_dxy_to_bars(df.index, dxy_df)
# Mirror process_pipeline: add USDCHF if the master file exists
usdchf_df = load_usdchf_data(USDCHF_MASTER_PATH)
if usdchf_df is not None:
df["usdchf_log_return"] = map_usdchf_to_bars(df.index, usdchf_df)
df.dropna(inplace=True)
return df