{ "nbformat": 4, "nbformat_minor": 0, "metadata": { "colab": { "provenance": [], "gpuType": "T4" }, "kernelspec": { "name": "python3", "display_name": "Python 3" }, "language_info": { "name": "python" }, "accelerator": "GPU" }, "cells": [ { "cell_type": "code", "execution_count": 12, "metadata": { "id": "N5p679fK85c7" }, "outputs": [], "source": [ "def setup_colab_environment():\n", " \"\"\"\n", " Sets up the Google Colab environment by installing dependencies,\n", " mounting Google Drive, and injecting a keep-alive script.\n", " \"\"\"\n", " print(\"--- Setting up Google Colab Environment ---\")\n", "\n", " # 1. Install required libraries\n", " print(\"\\n[1/3] Installing required libraries...\")\n", " # Using -q for a quieter installation\n", " !pip install -q python-dotenv numpy pandas matplotlib shap xgboost optuna requests scikit-learn pydantic\n", " # Install PyTorch and PyG for optional GNN features\n", " !pip install -q torch torch-geometric\n", "\n", " print(\"Libraries installed successfully.\")" ] }, { "cell_type": "code", "source": [ " # 2. Mount Google Drive\n", " print(\"\\n[2/3] Mounting Google Drive...\")\n", " try:\n", " from google.colab import drive\n", " drive.mount('/content/drive')\n", " print(\"Google Drive mounted successfully at /content/drive\")\n", " except Exception as e:\n", " print(f\"Error mounting Google Drive: {e}\")\n", " print(\"Please ensure you are running this in a Google Colab environment.\")" ], "metadata": { "id": "NESKAZZi9Dvd", "colab": { "base_uri": "https://localhost:8080/" }, "outputId": "0addcc85-d34c-45e8-9473-bb030aca8fe9" }, "execution_count": 13, "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "\n", "[2/3] Mounting Google Drive...\n", "Drive already mounted at /content/drive; to attempt to forcibly remount, call drive.mount(\"/content/drive\", force_remount=True).\n", "Google Drive mounted successfully at /content/drive\n" ] } ] }, { "cell_type": "code", "source": [ " # 3. Implement Colab keep-alive to prevent timeout\n", " print(\"\\n[3/3] Implementing Colab keep-alive function...\")\n", " try:\n", " from IPython.display import display, Javascript\n", "\n", " # This JavaScript function will periodically log to the console,\n", " # which tricks Colab into thinking the notebook is active.\n", " keep_alive_js = Javascript(\"\"\"\n", " function ColabKeepAlive() {\n", " console.log(\"Colab keep-alive running...\");\n", " setInterval(function() {\n", " console.log(\"Keeping session active...\");\n", " }, 300000); // 300000ms = 5 minutes\n", " }\n", " ColabKeepAlive();\n", " \"\"\")\n", " display(keep_alive_js)\n", " print(\"Keep-alive script injected. Notebook will resist disconnection.\")\n", " except Exception as e:\n", " print(f\"Could not inject keep-alive script: {e}\")\n", "\n", " print(\"\\n--- Environment setup complete. You can now run the main script. ---\")\n", " print(\"IMPORTANT: Make sure your 'TradingData' folder is in your Google Drive's main directory.\")\n", " print(\"If it's elsewhere, you MUST update the 'BASE_PATH' in the 'main()' function at the end of this script.\")" ], "metadata": { "colab": { "base_uri": "https://localhost:8080/", "height": 139 }, "id": "-E9KYV4lmJ69", "outputId": "c324aa8f-210a-423a-e84e-189a67fb4721" }, "execution_count": 14, "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "\n", "[3/3] Implementing Colab keep-alive function...\n" ] }, { "output_type": "display_data", "data": { "text/plain": [ "" ], "application/javascript": [ "\n", " function ColabKeepAlive() {\n", " console.log(\"Colab keep-alive running...\");\n", " setInterval(function() {\n", " console.log(\"Keeping session active...\");\n", " }, 300000); // 300000ms = 5 minutes\n", " }\n", " ColabKeepAlive();\n", " " ] }, "metadata": {} }, { "output_type": "stream", "name": "stdout", "text": [ "Keep-alive script injected. Notebook will resist disconnection.\n", "\n", "--- Environment setup complete. You can now run the main script. ---\n", "IMPORTANT: Make sure your 'TradingData' folder is in your Google Drive's main directory.\n", "If it's elsewhere, you MUST update the 'BASE_PATH' in the 'main()' function at the end of this script.\n" ] } ] }, { "cell_type": "code", "source": [ "# Run the setup function\n", "setup_colab_environment()" ], "metadata": { "colab": { "base_uri": "https://localhost:8080/" }, "id": "_-Pfh7GKmLg6", "outputId": "050c6dd5-beb3-4611-d788-d425e5174a8f" }, "execution_count": 15, "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "--- Setting up Google Colab Environment ---\n", "\n", "[1/3] Installing required libraries...\n", "Libraries installed successfully.\n" ] } ] }, { "cell_type": "code", "source": [ "# =============================================================================\n", "# MAIN TRADING FRAMEWORK SCRIPT\n", "# =============================================================================\n", "import os\n", "import re\n", "import json\n", "import time\n", "import warnings\n", "import logging\n", "import sys\n", "import random\n", "import shutil\n", "from datetime import datetime, date, timedelta\n", "from logging.handlers import RotatingFileHandler\n", "from typing import List, Dict, Any, Optional, Tuple, Union, Callable\n", "from collections import defaultdict\n", "import pathlib\n", "\n", "# --- LOAD ENVIRONMENT VARIABLES ---\n", "from dotenv import load_dotenv\n", "# The .env file will be loaded from the Google Drive path specified in main()\n", "# --- END ---\n", "\n", "import numpy as np\n", "import pandas as pd\n", "import matplotlib.pyplot as plt\n", "import shap\n", "import xgboost as xgb\n", "import optuna\n", "import requests\n", "from sklearn.model_selection import train_test_split\n", "from sklearn.metrics import f1_score\n", "from sklearn.pipeline import Pipeline\n", "from sklearn.preprocessing import RobustScaler, MinMaxScaler\n", "from sklearn.utils.class_weight import compute_class_weight\n", "from pydantic import BaseModel, DirectoryPath, confloat, conint, Field\n", "from sklearn.ensemble import IsolationForest\n", "\n", "\n", "# --- DIAGNOSTICS & LOGGING SETUP ---\n", "logger = logging.getLogger(\"ML_Trading_Framework\")\n", "\n", "# --- GNN Specific Imports (requires PyTorch, PyG) ---\n", "try:\n", " import torch\n", " import torch.nn.functional as F\n", " from torch_geometric.data import Data\n", " from torch_geometric.nn import GCNConv\n", " from torch.optim import Adam\n", " GNN_AVAILABLE = True\n", "except ImportError:\n", " GNN_AVAILABLE = False\n", " class GCNConv: pass\n", " class Adam: pass\n", " class Data: pass\n", " def F(): pass\n", " torch = None\n", "\n", "# --- LOGGING SWITCHES ---\n", "LOG_ANOMALY_SKIPS = False\n", "LOG_PARTIAL_PROFITS = True\n", "# -----------------------------\n", "\n", "def flush_loggers():\n", " \"\"\"Flushes all handlers for all active loggers to disk.\"\"\"\n", " for handler in logging.getLogger().handlers:\n", " handler.flush()\n", " for handler in logging.getLogger(\"ML_Trading_Framework\").handlers:\n", " handler.flush()\n", "\n", "def setup_logging() -> logging.Logger:\n", " if logger.hasHandlers():\n", " logger.handlers.clear()\n", " logger.setLevel(logging.DEBUG)\n", " ch = logging.StreamHandler(sys.stdout)\n", " ch.setLevel(logging.INFO)\n", " formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')\n", " ch.setFormatter(formatter)\n", " logger.addHandler(ch)\n", " if GNN_AVAILABLE:\n", " logger.info(\"PyTorch and PyG loaded successfully. GNN module is available.\")\n", " else:\n", " logger.warning(\"PyTorch or PyTorch Geometric not found. GNN-based strategies will be unavailable.\")\n", " return logger\n", "\n", "logger = setup_logging()\n", "optuna.logging.set_verbosity(optuna.logging.WARNING)\n", "# --- END DIAGNOSTICS & LOGGING ---\n", "\n", "\n", "warnings.filterwarnings('ignore', category=FutureWarning)\n", "warnings.filterwarnings('ignore', category=UserWarning)\n", "warnings.filterwarnings('ignore', category=pd.errors.PerformanceWarning)\n", "\n", "# =============================================================================\n", "# 2. GEMINI AI ANALYZER & API TIMER\n", "# =============================================================================\n", "class APITimer:\n", " \"\"\"Manages the timing of API calls to ensure a minimum interval between them.\"\"\"\n", " def __init__(self, interval_seconds: int = 300):\n", " self.interval = timedelta(seconds=interval_seconds)\n", " self.last_call_time: Optional[datetime] = None\n", " if self.interval.total_seconds() > 0:\n", " logger.info(f\"API Timer initialized with a {self.interval.total_seconds():.0f}-second interval.\")\n", " else:\n", " logger.info(\"API Timer initialized with a 0-second interval (timer is effectively disabled).\")\n", "\n", " def _wait_if_needed(self):\n", " if self.interval.total_seconds() <= 0: return\n", " if self.last_call_time is None: return\n", "\n", " elapsed = datetime.now() - self.last_call_time\n", " wait_time_delta = self.interval - elapsed\n", " wait_seconds = wait_time_delta.total_seconds()\n", "\n", " if wait_seconds > 0:\n", " logger.info(f\" - Time since last API call: {elapsed.total_seconds():.1f} seconds.\")\n", " logger.info(f\" - Waiting for {wait_seconds:.1f} seconds to respect the {self.interval.total_seconds():.0f}s interval...\")\n", " flush_loggers()\n", " time.sleep(wait_seconds)\n", " else:\n", " logger.info(f\" - Time since last API call ({elapsed.total_seconds():.1f}s) exceeds interval. No wait needed.\")\n", "\n", " def call(self, api_function: Callable, *args, **kwargs) -> Any:\n", " \"\"\"Executes the API function after ensuring the timing interval is met.\"\"\"\n", " self._wait_if_needed()\n", " self.last_call_time = datetime.now()\n", " logger.info(f\" - Making API call to '{api_function.__name__}' at {self.last_call_time.strftime('%H:%M:%S')}...\")\n", " result = api_function(*args, **kwargs)\n", " logger.info(f\" - API call to '{api_function.__name__}' complete.\")\n", " return result\n", "\n", "class GeminiAnalyzer:\n", " def __init__(self):\n", " api_key = os.getenv(\"GEMINI_API_KEY\")\n", " if not api_key or \"YOUR\" in api_key or \"PASTE\" in api_key:\n", " logger.warning(\"!CRITICAL! GEMINI_API_KEY not found in environment or is a placeholder.\")\n", " try:\n", " # In Colab, input() works in the cell's output area.\n", " api_key = input(\">>> Please paste your Gemini API Key and press Enter, or press Enter to skip: \").strip()\n", " if not api_key:\n", " logger.warning(\"No API Key provided. AI analysis will be skipped.\")\n", " self.api_key_valid = False\n", " else:\n", " logger.info(\"Using API Key provided via manual input.\")\n", " self.api_key_valid = True\n", " except Exception:\n", " logger.warning(\"Could not read input. AI analysis will be skipped.\")\n", " self.api_key_valid = False\n", " api_key = None\n", " else:\n", " logger.info(\"Successfully loaded GEMINI_API_KEY from environment.\")\n", " self.api_key_valid = True\n", "\n", " if self.api_key_valid:\n", " self.api_url = f\"https://generativelanguage.googleapis.com/v1beta/models/gemini-2.0-flash:generateContent?key={api_key}\"\n", " self.headers = {\"Content-Type\": \"application/json\"}\n", " else:\n", " self.api_url = \"\"\n", " self.headers = {}\n", "\n", " def _sanitize_value(self, value: Any) -> Any:\n", " if isinstance(value, pathlib.Path): return str(value)\n", " if isinstance(value, (np.int64, np.int32)): return int(value)\n", " if isinstance(value, (np.float64, np.float32)):\n", " if np.isnan(value) or np.isinf(value): return None\n", " return float(value)\n", " if isinstance(value, (pd.Timestamp, datetime, date)): return value.isoformat()\n", " return value\n", "\n", " def _sanitize_dict(self, data: Any) -> Any:\n", " if isinstance(data, dict): return {key: self._sanitize_dict(value) for key, value in data.items()}\n", " if isinstance(data, list): return [self._sanitize_dict(item) for item in data]\n", " return self._sanitize_value(data)\n", "\n", " def _call_gemini(self, prompt: str) -> str:\n", " if not self.api_key_valid: return \"{}\"\n", " if len(prompt) > 30000: logger.warning(\"Prompt is very large, may risk exceeding token limits.\")\n", " payload = {\"contents\": [{\"parts\": [{\"text\": prompt}]}]}\n", " sanitized_payload = self._sanitize_dict(payload)\n", "\n", " retry_delays = [5, 15, 30]\n", "\n", " for attempt, delay in enumerate([0] + retry_delays):\n", " if delay > 0:\n", " logger.warning(f\"API connection failed. Retrying in {delay} seconds... (Attempt {attempt}/{len(retry_delays)})\")\n", " flush_loggers()\n", " time.sleep(delay)\n", "\n", " try:\n", " response = requests.post(self.api_url, headers=self.headers, data=json.dumps(sanitized_payload), timeout=120)\n", " response.raise_for_status()\n", "\n", " result = response.json()\n", " if \"candidates\" in result and result[\"candidates\"] and \"content\" in result[\"candidates\"][0] and \"parts\" in result[\"candidates\"][0][\"content\"]:\n", " return result[\"candidates\"][0][\"content\"][\"parts\"][0][\"text\"]\n", " else:\n", " logger.error(f\"Invalid Gemini response structure: {result}\")\n", " return \"{}\"\n", "\n", " except requests.exceptions.RequestException as e:\n", " logger.error(f\"Gemini API request failed on attempt {attempt + 1}: {e}\")\n", " if attempt == len(retry_delays):\n", " logger.critical(\"API connection failed after all retries. Stopping.\")\n", " return \"{}\"\n", " except json.JSONDecodeError as e:\n", " logger.error(f\"Failed to decode Gemini response JSON: {e} - Response: {response.text}\")\n", " return \"{}\"\n", " except (KeyError, IndexError) as e:\n", " logger.error(f\"Failed to extract text from Gemini response: {e} - Response: {response.text}\")\n", " return \"{}\"\n", "\n", " return \"{}\"\n", "\n", " def _extract_json_from_response(self, response_text: str) -> Dict:\n", " if response_text.strip().lower() == 'null':\n", " return {}\n", "\n", " try:\n", " match = re.search(r\"```json\\s*(.*?)\\s*```\", response_text, re.DOTALL)\n", " json_text = match.group(1) if match else response_text\n", "\n", " if json_text.strip().lower() == 'null':\n", " return {}\n", "\n", " suggestions = json.loads(json_text.strip())\n", "\n", " if not isinstance(suggestions, dict):\n", " logger.error(f\"Parsed JSON is not a dictionary. Response text: {response_text}\")\n", " return {}\n", "\n", " if 'current_params' in suggestions and isinstance(suggestions.get('current_params'), dict):\n", " nested_params = suggestions.pop('current_params')\n", " suggestions.update(nested_params)\n", " return suggestions\n", " except (json.JSONDecodeError, AttributeError) as e:\n", " logger.error(f\"Could not parse JSON from response: {e}\\nResponse text: {response_text}\")\n", " return {}\n", "\n", " def get_initial_run_setup(self, script_version: str, ledger: Dict, memory: Dict, playbook: Dict, health_report: Dict, directives: List[Dict], data_summary: Dict) -> Dict:\n", " if not self.api_key_valid:\n", " logger.warning(\"No API key. Skipping AI-driven setup and using default config.\")\n", " return {}\n", "\n", " logger.info(\"-> Performing Initial AI Analysis & Setup (Single API Call)...\")\n", "\n", " nickname_prompt_part = \"\"\n", " if script_version not in ledger:\n", " theme = random.choice([\"Astronomical Objects\", \"Mythological Figures\", \"Gemstones\", \"Constellations\", \"Legendary Swords\"])\n", " nickname_prompt_part = (\n", " \"1. **Generate Nickname**: The script version is new. Generate a unique, cool-sounding, one-word codename for this run. \"\n", " f\"The theme is **{theme}**. Do not use any from this list of past names: {list(ledger.values())}. \"\n", " \"Place the new name in the `nickname` key of your JSON response.\\n\"\n", " )\n", " else:\n", " nickname_prompt_part = \"1. **Generate Nickname**: A nickname for this script version already exists. You do not need to generate a new one. Set the `nickname` key in your response to `null`.\\n\"\n", "\n", " directive_str = \"No specific directives for this run.\"\n", " if directives:\n", " directive_str = \"**CRITICAL DIRECTIVES FOR THIS RUN:**\\n\" + \"\\n\".join(f\"- {d.get('reason', 'Unnamed directive')}\" for d in directives)\n", "\n", " health_report_str = \"No long-term health report available.\"\n", " if health_report:\n", " health_report_str = f\"**STRATEGIC HEALTH ANALYSIS (Lower scores are better):**\\n{json.dumps(health_report, indent=2)}\\n\\n\"\n", "\n", " if not GNN_AVAILABLE:\n", " playbook = {k: v for k, v in playbook.items() if not v.get(\"requires_gnn\")}\n", " logger.warning(\"GNN strategies filtered from playbook due to missing libraries.\")\n", "\n", " context_feature_mandate = (\"- `selected_features`: This key is **MANDATORY**. You must provide a non-empty list. Start by using the default features for your chosen strategy from the playbook. **CRITICAL: You MUST ensure the list includes at least TWO multi-timeframe context features** (e.g., `DAILY_ctx_Trend`, `H1_ctx_SMA`) unless the strategy is GNN-based.\")\n", "\n", " prompt = (\n", " \"You are a Master Trading Strategist responsible for configuring a trading framework for its next run.\\n\\n\"\n", " \"**YOUR THREE TASKS:**\\n\"\n", " f\"{nickname_prompt_part}\"\n", " \"2. **Analyze Market Data**: Review the `MARKET DATA SUMMARY` and comment on conditions in the `analysis_notes` key.\\n\"\n", " \"3. **Select Strategy & Configure**: Based on your analysis, select the optimal `strategy_name` from the playbook. Then, provide a complete starting configuration. **This is your most important task.** Your configuration MUST include a `selected_features` key. Start with the list of features provided for your chosen strategy in the playbook; you can make minor adjustments if needed. The list must not be empty unless the strategy requires GNN.\\n\\n\"\n", " \"**PARAMETER RULES & GUIDANCE:**\\n\"\n", " \"- `strategy_name`: MUST be one of the keys from the playbook.\\n\"\n", " f\"{context_feature_mandate}\\n\"\n", " \"- `RETRAINING_FREQUENCY`: **MUST be a string** with a number and unit (e.g., '90D', '8W', '2M').\\n\"\n", " \"- `USE_PARTIAL_PROFIT`: **MUST be a boolean (true/false).** Start with `false` unless you have a strong reason to enable it.\\n\"\n", " \"- If `USE_PARTIAL_PROFIT` is true, provide `PARTIAL_PROFIT_TRIGGER_R` (float) and `PARTIAL_PROFIT_TAKE_PCT` (float).\\n\"\n", " \"- `MAX_CONCURRENT_TRADES`: An integer from 1 to 10.\\n\"\n", " \"- `MAX_DD_PER_CYCLE`: A float between 0.05 and 0.5.\\n\\n\"\n", " \"**OUTPUT FORMAT**: Respond ONLY with a single, valid JSON object containing all required keys.\\n\\n\"\n", " \"--- CONTEXT FOR YOUR DECISION ---\\n\\n\"\n", " f\"**1. MARKET DATA SUMMARY:**\\n{json.dumps(self._sanitize_dict(data_summary), indent=2)}\\n\\n\"\n", " f\"**2. CRITICAL DIRECTIVES:**\\n{directive_str}\\n\\n\"\n", " f\"**3. STRATEGIC HEALTH & MEMORY:**\\n{health_report_str}\"\n", " f\"Framework Memory (Champion & History):\\n{json.dumps(self._sanitize_dict(memory), indent=2)}\\n\\n\"\n", " f\"**4. STRATEGY PLAYBOOK (Your options):**\\n{json.dumps(playbook, indent=2)}\\n\"\n", " )\n", "\n", "\n", " response_text = self._call_gemini(prompt)\n", " suggestions = self._extract_json_from_response(response_text)\n", "\n", " if suggestions and \"strategy_name\" in suggestions:\n", " logger.info(\" - Initial AI Analysis and Setup complete.\")\n", " return suggestions\n", " else:\n", " logger.error(\" - AI-driven setup failed to return a valid configuration.\")\n", " return {}\n", "\n", " def analyze_cycle_and_suggest_changes(self, historical_results: List[Dict], available_features: List[str], strategy_details: Dict, cycle_status: str) -> Dict:\n", " if not self.api_key_valid: return {}\n", "\n", " base_prompt_intro = \"You are an expert trading model analyst. Your primary goal is to create a STABLE and PROFITABLE strategy by tuning its parameters within a walk-forward analysis.\"\n", " json_response_format = \"Respond ONLY with a valid JSON object containing your suggested keys.\"\n", " context_feature_mandate = (\"- `selected_features`: This key is **MANDATORY**. You must provide a non-empty list. Start by using the available features. **CRITICAL: You MUST ensure the list includes at least TWO multi-timeframe context features** (e.g., `DAILY_ctx_Trend`, `H1_ctx_SMA`) unless the strategy is GNN-based.\")\n", "\n", " if cycle_status == \"TRAINING_FAILURE\":\n", " prompt_details = (\n", " \"**CRITICAL: MODEL TRAINING FAILURE!**\\n\"\n", " \"The previous training attempt resulted in a model with an unacceptably low quality score. The model was discarded before backtesting.\\n\\n\"\n", " \"**YOUR TASK:**\\n\"\n", " \"Propose a **significant change** to the model's configuration to fix the training issue. This is not a time for small tweaks.\\n\"\n", " \"1. **Change `selected_features`:** The current feature set is likely poor. Try a different combination. **You must still include context features.**\\n\"\n", " \"2. **Adjust `OPTUNA_TRIALS` or `LOOKAHEAD_CANDLES`:** The optimization or labeling process may be flawed. Suggest new values.\\n\"\n", " )\n", " elif cycle_status == \"PROBATION\":\n", " prompt_details = (\n", " \"**STRATEGY ON PROBATION**\\n\"\n", " \"The previous trading cycle for this strategy hit its maximum drawdown ('Circuit Breaker'). Your primary goal is to **REDUCE RISK** while trying to maintain profitability.\\n\\n\"\n", " \"**YOUR TASK (Safety First):**\\n\"\n", " \"1. **You MUST suggest changes that lower risk.** This is not optional.\\n\"\n", " \"2. **Lower risk by:** reducing `MAX_DD_PER_CYCLE` (must be between 0.05-0.20), reducing `MAX_CONCURRENT_TRADES`.\\n\"\n", " \"3. Consider a shorter `RETRAINING_FREQUENCY` for faster adaptation.\\n\"\n", " f\"4. {context_feature_mandate}\\n\"\n", " )\n", " else: # Standard cycle\n", " prompt_details = (\n", " \"**STANDARD CYCLE REVIEW**\\n\"\n", " \"Review the recent cycle performance and suggest a new configuration to improve the model.\\n\\n\"\n", " \"**YOUR TASK:**\\n\"\n", " \"Suggest a new configuration for any of the following parameters:\\n\"\n", " \"- `MAX_DD_PER_CYCLE`, `RETRAINING_FREQUENCY`, `MAX_CONCURRENT_TRADES`, `USE_PARTIAL_PROFIT`, etc.\\n\"\n", " f\"- {context_feature_mandate}\\n\"\n", " )\n", "\n", " data_context = (\n", " f\"**SUMMARIZED HISTORICAL CYCLE RESULTS:**\\n{json.dumps(self._sanitize_dict(historical_results), indent=2)}\\n\\n\"\n", " f\"**AVAILABLE FEATURES FOR THIS STRATEGY:**\\n{available_features}\"\n", " )\n", "\n", " prompt = f\"{base_prompt_intro}\\n\\n{prompt_details}\\n\\n{json_response_format}\\n\\n{data_context}\"\n", "\n", " response_text = self._call_gemini(prompt)\n", " suggestions = self._extract_json_from_response(response_text)\n", " return suggestions\n", "\n", " def propose_strategic_intervention(self, failure_history: List[Dict], playbook: Dict, last_failed_strategy: str, quarantine_list: List[str]) -> Dict:\n", " if not self.api_key_valid: return {}\n", " logger.warning(\"! STRATEGIC INTERVENTION !: Current strategy has failed repeatedly. Engaging AI to select a new strategy.\")\n", "\n", " available_playbook = { k: v for k, v in playbook.items() if k not in quarantine_list and (GNN_AVAILABLE or not v.get(\"requires_gnn\"))}\n", " context_feature_mandate = (\"**You MUST provide a `selected_features` list. Start with the default features from the playbook for the new strategy you choose. The list MUST include at least TWO multi-timeframe context features** (e.g., `DAILY_ctx_Trend`, `H1_ctx_SMA`).\")\n", "\n", " prompt = (\n", " \"You are a master strategist executing an emergency intervention. The current strategy, \"\n", " f\"**`{last_failed_strategy}`**, has failed multiple consecutive cycles, even after being given a chance on probation. It is now being quarantined.\\n\\n\"\n", " \"**CRITICAL INSTRUCTIONS:**\\n\"\n", " f\"1. **CRITICAL CONSTRAINT:** The following strategies are in 'quarantine' due to recent, repeated failures. **YOU MUST NOT SELECT ANY STRATEGY FROM THIS LIST: {quarantine_list}**\\n\"\n", " \"2. **Select a NEW STRATEGY:** You **MUST** choose a *different* strategy from the available playbook that is NOT in the quarantine list.\\n\"\n", " f\"3. **Propose a Safe Starting Configuration:** Provide a reasonable and SAFE starting configuration for this new strategy. {context_feature_mandate} Start with conservative values: `RETRAINING_FREQUENCY`: '90D', `MAX_DD_PER_CYCLE`: 0.15 (float), `MAX_CONCURRENT_TRADES`: 2, and **`USE_PARTIAL_PROFIT`: false**.\\n\\n\"\n", " f\"**RECENT FAILED HISTORY (for context):**\\n{json.dumps(self._sanitize_dict(failure_history), indent=2)}\\n\\n\"\n", " f\"**AVAILABLE STRATEGIES (PLAYBOOK):**\\n{json.dumps(available_playbook, indent=2)}\\n\\n\"\n", " \"Respond ONLY with a valid JSON object for the new configuration, including `strategy_name` and `selected_features`.\"\n", " )\n", "\n", " response_text = self._call_gemini(prompt)\n", " suggestions = self._extract_json_from_response(response_text)\n", " return suggestions\n", "\n", " def propose_regime_based_strategy_switch(self, regime_data_summary: Dict, playbook: Dict, current_strategy_name: str, quarantine_list: List[str]) -> Dict:\n", " \"\"\"New AI endpoint for pre-cycle regime analysis.\"\"\"\n", " if not self.api_key_valid: return {}\n", "\n", " logger.info(\" - Performing Pre-Cycle Regime Analysis...\")\n", "\n", " available_playbook = {k: v for k, v in playbook.items() if k not in quarantine_list}\n", "\n", " prompt = (\n", " \"You are a market regime analyst. The framework is about to start a new walk-forward cycle.\\n\\n\"\n", " \"**YOUR TASK:**\\n\"\n", " f\"The framework is currently configured to use the **`{current_strategy_name}`** strategy. Based on the `RECENT MARKET DATA SUMMARY` provided below, decide if this is still the optimal choice.\\n\\n\"\n", " \"1. **Analyze the Data**: Review the `average_adx`, `volatility_rank`, and `trending_percentage` to diagnose the current market regime (e.g., strong trend, weak trend, ranging, volatile, quiet).\\n\"\n", " \"2. **Review the Playbook**: Compare your diagnosis with the intended purpose of the strategies in the `STRATEGY PLAYBOOK`.\\n\"\n", " \"3. **Make a Decision**:\\n\"\n", " \" - If you believe a **different strategy is better suited** to the current market regime, respond with the JSON configuration for that new strategy (just the strategy name and its default features/params from the playbook).\\n\"\n", " \" - If you believe the **current strategy remains the best fit**, respond with `null`.\\n\\n\"\n", " \"**RESPONSE FORMAT**: Respond ONLY with the JSON for the new strategy OR the word `null`.\\n\\n\"\n", " \"--- CONTEXT FOR YOUR DECISION ---\\n\\n\"\n", " f\"**1. RECENT MARKET DATA SUMMARY (Last ~30 Days):**\\n{json.dumps(regime_data_summary, indent=2)}\\n\\n\"\n", " f\"**2. STRATEGY PLAYBOOK (Your options):**\\n{json.dumps(available_playbook, indent=2)}\\n\"\n", " )\n", "\n", " response_text = self._call_gemini(prompt)\n", "\n", " if response_text.strip().lower() == 'null':\n", " logger.info(\" - AI analysis confirms current strategy is optimal for the upcoming regime. No changes made.\")\n", " return {}\n", "\n", " suggestions = self._extract_json_from_response(response_text)\n", " return suggestions\n", "\n", "# =============================================================================\n", "# 3. CONFIGURATION & VALIDATION\n", "# =============================================================================\n", "class ConfigModel(BaseModel):\n", " BASE_PATH: DirectoryPath; REPORT_LABEL: str; INITIAL_CAPITAL: confloat(gt=0)\n", " CONFIDENCE_TIERS: Dict[str, Dict[str, Any]]; BASE_RISK_PER_TRADE_PCT: confloat(gt=0, lt=1)\n", " SPREAD_PCTG_OF_ATR: confloat(ge=0); SLIPPAGE_PCTG_OF_ATR: confloat(ge=0)\n", " MAX_DD_PER_CYCLE: confloat(ge=0.05, lt=1.0) = 0.25; RISK_CAP_PER_TRADE_USD: confloat(gt=0)\n", " OPTUNA_TRIALS: conint(gt=0); TRAINING_WINDOW: str; RETRAINING_FREQUENCY: str\n", " FORWARD_TEST_GAP: str; LOOKAHEAD_CANDLES: conint(gt=0); CALCULATE_SHAP_VALUES: bool = True\n", " TREND_FILTER_THRESHOLD: confloat(gt=0) = 25.0; BOLLINGER_PERIOD: conint(gt=0) = 20\n", " STOCHASTIC_PERIOD: conint(gt=0) = 14; GNN_EMBEDDING_DIM: conint(gt=0) = 8\n", " GNN_EPOCHS: conint(gt=0) = 50; MIN_VOLATILITY_RANK: confloat(ge=0.0, le=1.0) = 0.1\n", " MAX_VOLATILITY_RANK: confloat(ge=0.0, le=1.0) = 0.9; selected_features: List[str]\n", " run_timestamp: str; strategy_name: str; nickname: str = \"\"\n", " analysis_notes: str = \"\"\n", " MAX_CONCURRENT_TRADES: conint(ge=1, le=20) = 3\n", " USE_PARTIAL_PROFIT: bool = False\n", " PARTIAL_PROFIT_TRIGGER_R: confloat(gt=0) = 1.5\n", " PARTIAL_PROFIT_TAKE_PCT: confloat(ge=0.1, le=0.9) = 0.5\n", " MAX_TRAINING_RETRIES_PER_CYCLE: conint(ge=0) = 3\n", "\n", " MODEL_SAVE_PATH: str = Field(default=\"\", repr=False); PLOT_SAVE_PATH: str = Field(default=\"\", repr=False)\n", " REPORT_SAVE_PATH: str = Field(default=\"\", repr=False); SHAP_PLOT_PATH: str = Field(default=\"\", repr=False)\n", " LOG_FILE_PATH: str = Field(default=\"\", repr=False); CHAMPION_FILE_PATH: str = Field(default=\"\", repr=False)\n", " HISTORY_FILE_PATH: str = Field(default=\"\", repr=False); PLAYBOOK_FILE_PATH: str = Field(default=\"\", repr=False)\n", " DIRECTIVES_FILE_PATH: str = Field(default=\"\", repr=False); NICKNAME_LEDGER_PATH: str = Field(default=\"\", repr=False)\n", "\n", " def __init__(self, **data: Any):\n", " super().__init__(**data)\n", " # All paths now point to the local workspace BASE_PATH\n", " results_dir = os.path.join(self.BASE_PATH, \"Results\")\n", " version_match = re.search(r'V(\\d+)', self.REPORT_LABEL)\n", " version_str = f\"_V{version_match.group(1)}\" if version_match else \"\"\n", " folder_name = f\"{self.nickname}{version_str}\" if self.nickname and version_str else self.REPORT_LABEL\n", " run_id = f\"{folder_name}_{self.strategy_name}_{self.run_timestamp}\"\n", " result_folder_path = os.path.join(results_dir, folder_name)\n", "\n", " if self.nickname and self.nickname != \"init\":\n", " os.makedirs(result_folder_path, exist_ok=True)\n", "\n", " self.MODEL_SAVE_PATH = os.path.join(result_folder_path, f\"{run_id}_model.json\")\n", " self.PLOT_SAVE_PATH = os.path.join(result_folder_path, f\"{run_id}_equity_curve.png\")\n", " self.REPORT_SAVE_PATH = os.path.join(result_folder_path, f\"{run_id}_report.txt\")\n", " self.SHAP_PLOT_PATH = os.path.join(result_folder_path, f\"{run_id}_shap_summary.png\")\n", " self.LOG_FILE_PATH = os.path.join(result_folder_path, f\"{run_id}_run.log\")\n", "\n", " # Core state files also point to the local workspace\n", " self.CHAMPION_FILE_PATH = os.path.join(results_dir, \"champion.json\")\n", " self.HISTORY_FILE_PATH = os.path.join(results_dir, \"historical_runs.jsonl\")\n", " self.PLAYBOOK_FILE_PATH = os.path.join(results_dir, \"strategy_playbook.json\")\n", " self.DIRECTIVES_FILE_PATH = os.path.join(results_dir, \"framework_directives.json\")\n", " self.NICKNAME_LEDGER_PATH = os.path.join(results_dir, \"nickname_ledger.json\")\n", "\n", "# =============================================================================\n", "# 4. DATA LOADER & 5. FEATURE ENGINEERING\n", "# =============================================================================\n", "class DataLoader:\n", " def __init__(self, config: ConfigModel, data_source_path: str):\n", " self.config = config\n", " # Data is loaded from the persistent Drive path, not the temp workspace\n", " self.data_source_path = data_source_path\n", "\n", " def _parse_single_file(self, file_path: str, filename: str) -> Optional[pd.DataFrame]:\n", " try:\n", " parts = filename.split('_'); symbol, tf = parts[0], parts[1]\n", " df = pd.read_csv(file_path, delimiter='\\t' if '\\t' in open(file_path, encoding='utf-8').readline() else ',')\n", " df.columns = [c.upper().replace('<', '').replace('>', '') for c in df.columns]\n", " date_col = next((c for c in df.columns if 'DATE' in c), None)\n", " time_col = next((c for c in df.columns if 'TIME' in c), None)\n", " if date_col and time_col: df['Timestamp'] = pd.to_datetime(df[date_col] + ' ' + df[time_col], errors='coerce')\n", " elif date_col: df['Timestamp'] = pd.to_datetime(df[date_col], errors='coerce')\n", " else: logger.error(f\" - No date/time columns found in {filename}.\"); return None\n", " df.dropna(subset=['Timestamp'], inplace=True); df.set_index('Timestamp', inplace=True)\n", " col_map = {c: c.capitalize() for c in df.columns if c.lower() in ['open', 'high', 'low', 'close', 'tickvol', 'volume', 'spread']}\n", " df.rename(columns=col_map, inplace=True)\n", " vol_col = 'Volume' if 'Volume' in df.columns else 'Tickvol'\n", " df.rename(columns={vol_col: 'RealVolume'}, inplace=True, errors='ignore')\n", " if 'RealVolume' not in df.columns: df['RealVolume'] = 0\n", " df['RealVolume'] = pd.to_numeric(df['RealVolume'], errors='coerce').fillna(0)\n", " df['Symbol'] = symbol; return df\n", " except Exception as e: logger.error(f\" - Failed to load {filename}: {e}\", exc_info=True); return None\n", "\n", " def load_and_parse_data(self, filenames: List[str]) -> Tuple[Optional[Dict[str, pd.DataFrame]], List[str]]:\n", " logger.info(\"-> Stage 1: Loading and Preparing Multi-Timeframe Data...\")\n", " data_by_tf = defaultdict(list)\n", " for filename in filenames:\n", " file_path = os.path.join(self.data_source_path, filename)\n", " if not os.path.exists(file_path): logger.warning(f\" - File not found, skipping: {file_path}\"); continue\n", " df = self._parse_single_file(file_path, filename)\n", " if df is not None: tf = filename.split('_')[1]; data_by_tf[tf].append(df)\n", " processed_dfs: Dict[str, pd.DataFrame] = {}\n", " for tf, dfs in data_by_tf.items():\n", " if dfs:\n", " combined = pd.concat(dfs)\n", " all_symbols_df = [df[~df.index.duplicated(keep='first')].sort_index() for _, df in combined.groupby('Symbol')]\n", " final_combined = pd.concat(all_symbols_df).sort_index()\n", " processed_dfs[tf] = final_combined\n", " logger.info(f\" - Processed {tf}: {len(final_combined):,} rows for {len(final_combined['Symbol'].unique())} symbols.\")\n", " detected_timeframes = list(processed_dfs.keys())\n", " if not processed_dfs: logger.critical(\" - Data loading failed for all files.\"); return None, []\n", " logger.info(f\"[SUCCESS] Data loading complete. Detected timeframes: {detected_timeframes}\")\n", " return processed_dfs, detected_timeframes\n", "\n", "class FeatureEngineer:\n", " TIMEFRAME_MAP = {'M1': 1,'M5': 5,'M15': 15,'M30': 30,'H1': 60,'H4': 240,'D1': 1440, 'DAILY': 1440}\n", " ANOMALY_FEATURES = ['ATR', 'bollinger_bandwidth', 'RSI', 'RealVolume', 'candle_body_size']\n", "\n", " def __init__(self, config: ConfigModel, timeframe_roles: Dict[str, str]):\n", " self.config = config\n", " self.roles = timeframe_roles\n", "\n", " def _get_weights_ffd(self, d: float, thres: float) -> np.ndarray:\n", " w, k = [1.], 1\n", " while True:\n", " w_ = -w[-1] / k * (d - k + 1)\n", " if abs(w_) < thres: break\n", " w.append(w_)\n", " k += 1\n", " return np.array(w[::-1]).reshape(-1, 1)\n", "\n", " def _fractional_differentiation(self, series: pd.Series, d: float, thres: float = 1e-5) -> pd.Series:\n", " weights = self._get_weights_ffd(d, thres)\n", " width = len(weights)\n", " if width > len(series): return pd.Series(index=series.index)\n", "\n", " diff_series = series.rolling(width).apply(lambda x: np.dot(weights.T, x)[0], raw=True)\n", " diff_series.name = f\"{series.name}_fracdiff_{d}\"\n", " return diff_series\n", "\n", " def _get_anomaly_scores(self, df: pd.DataFrame, features_to_check: list, contamination: float = 0.01) -> pd.Series:\n", " df_clean = df[features_to_check].dropna()\n", " if df_clean.empty:\n", " return pd.Series(1, index=df.index, name='anomaly_score')\n", "\n", " model = IsolationForest(contamination=contamination, random_state=42, n_estimators=100)\n", " model.fit(df_clean)\n", "\n", " scores = pd.Series(model.predict(df[features_to_check].fillna(0)), index=df.index)\n", " scores.name = 'anomaly_score'\n", " return scores\n", "\n", " def _calculate_adx(self, g:pd.DataFrame, period:int) -> pd.DataFrame:\n", " df=g.copy();alpha=1/period;df['tr']=pd.concat([df['High']-df['Low'],abs(df['High']-df['Close'].shift()),abs(df['Low']-df['Close'].shift())],axis=1).max(axis=1)\n", " df['dm_plus']=((df['High']-df['High'].shift())>(df['Low'].shift()-df['Low'])).astype(int)*(df['High']-df['High'].shift()).clip(lower=0)\n", " df['dm_minus']=((df['Low'].shift()-df['Low'])>(df['High']-df['High'].shift())).astype(int)*(df['Low'].shift()-df['Low']).clip(lower=0)\n", " atr_adx=df['tr'].ewm(alpha=alpha,adjust=False).mean();di_plus=100*(df['dm_plus'].ewm(alpha=alpha,adjust=False).mean()/atr_adx.replace(0,1e-9))\n", " di_minus=100*(df['dm_minus'].ewm(alpha=alpha,adjust=False).mean()/atr_adx.replace(0,1e-9));dx=100*(abs(di_plus-di_minus)/(di_plus+di_minus).replace(0,1e-9))\n", " g['ADX']=dx.ewm(alpha=alpha,adjust=False).mean();return g\n", "\n", " def _calculate_bollinger_bands(self, g:pd.DataFrame, period:int) -> pd.DataFrame:\n", " rolling_close=g['Close'].rolling(window=period);middle_band=rolling_close.mean();std_dev=rolling_close.std()\n", " g['bollinger_upper'] = middle_band + (std_dev * 2); g['bollinger_lower'] = middle_band - (std_dev * 2)\n", " g['bollinger_bandwidth'] = (g['bollinger_upper'] - g['bollinger_lower']) / middle_band.replace(0,np.nan); return g\n", "\n", " def _calculate_stochastic(self, g:pd.DataFrame, period:int) -> pd.DataFrame:\n", " low_min=g['Low'].rolling(window=period).min();high_max=g['High'].rolling(window=period).max()\n", " g['stoch_k']=100*(g['Close']-low_min)/(high_max-low_min).replace(0,np.nan);g['stoch_d']=g['stoch_k'].rolling(window=3).mean();return g\n", "\n", " def _calculate_momentum(self, g:pd.DataFrame) -> pd.DataFrame:\n", " g['momentum_10'] = g['Close'].diff(10)\n", " g['momentum_20'] = g['Close'].diff(20)\n", " return g\n", "\n", " def _calculate_seasonality(self, g: pd.DataFrame) -> pd.DataFrame:\n", " g['month'] = g.index.month\n", " g['week_of_year'] = g.index.isocalendar().week.astype(int)\n", " g['day_of_month'] = g.index.day\n", " return g\n", "\n", " def _calculate_candlestick_patterns(self, g: pd.DataFrame) -> pd.DataFrame:\n", " g['candle_body_size'] = abs(g['Close'] - g['Open'])\n", " g['is_doji'] = (g['candle_body_size'] / g['ATR'].replace(0,1)).lt(0.1).astype(int)\n", " g['is_engulfing'] = ((g['candle_body_size'] > abs(g['Close'].shift() - g['Open'].shift())) & (np.sign(g['Close']-g['Open']) != np.sign(g['Close'].shift()-g['Open'].shift()))).astype(int)\n", " return g\n", "\n", " def _calculate_htf_features(self,df:pd.DataFrame,p:str,s:int,a:int)->pd.DataFrame:\n", " tf_id = p.upper()\n", " logger.info(f\" - Calculating HTF features for {tf_id}...\");results=[]\n", " for symbol,group in df.groupby('Symbol'):\n", " g=group.copy();sma=g['Close'].rolling(s,min_periods=s).mean();atr=(g['High']-g['Low']).rolling(a,min_periods=a).mean();trend=np.sign(g['Close']-sma)\n", " temp_df=pd.DataFrame(index=g.index)\n", " temp_df[f'{tf_id}_ctx_SMA']=sma\n", " temp_df[f'{tf_id}_ctx_ATR']=atr\n", " temp_df[f'{tf_id}_ctx_Trend']=trend\n", " shifted_df=temp_df.shift(1);shifted_df['Symbol']=symbol;results.append(shifted_df)\n", " return pd.concat(results).reset_index()\n", "\n", " def _calculate_base_tf_native(self, g:pd.DataFrame)->pd.DataFrame:\n", " g_out=g.copy();lookback=14\n", " g_out['ATR']=(g['High']-g['Low']).rolling(lookback).mean();delta=g['Close'].diff();gain=delta.where(delta > 0,0).ewm(com=lookback-1,adjust=False).mean()\n", " loss=-delta.where(delta < 0,0).ewm(com=lookback-1,adjust=False).mean();g_out['RSI']=100-(100/(1+(gain/loss.replace(0,1e-9))))\n", " g_out=self._calculate_adx(g_out,lookback)\n", " g_out=self._calculate_bollinger_bands(g_out,self.config.BOLLINGER_PERIOD)\n", " g_out=self._calculate_stochastic(g_out,self.config.STOCHASTIC_PERIOD)\n", " g_out = self._calculate_momentum(g_out)\n", " g_out = self._calculate_seasonality(g_out)\n", " g_out = self._calculate_candlestick_patterns(g_out)\n", " g_out['hour'] = g_out.index.hour;g_out['day_of_week'] = g_out.index.dayofweek\n", " g_out['market_regime']=np.where(g_out['ADX']>self.config.TREND_FILTER_THRESHOLD,1,0)\n", " sma_fast = g_out['Close'].rolling(window=20).mean(); sma_slow = g_out['Close'].rolling(window=50).mean()\n", " signal_series = pd.Series(np.where(sma_fast > sma_slow, 1.0, -1.0), index=g_out.index)\n", " g_out['primary_model_signal'] = signal_series.diff().fillna(0)\n", " g_out['market_volatility_index'] = g_out['ATR'].rolling(100).rank(pct=True)\n", " g_out['close_fracdiff'] = self._fractional_differentiation(g_out['Close'], d=0.5)\n", " return g_out\n", "\n", " def create_feature_stack(self, data_by_tf: Dict[str, pd.DataFrame]) -> pd.DataFrame:\n", " logger.info(\"-> Stage 2: Engineering Features from Multi-Timeframe Data...\")\n", "\n", " base_tf, medium_tf, high_tf = self.roles['base'], self.roles['medium'], self.roles['high']\n", " if base_tf not in data_by_tf:\n", " logger.critical(f\"Base timeframe '{base_tf}' data is missing. Cannot proceed.\"); return pd.DataFrame()\n", "\n", " df_base_list = [self._calculate_base_tf_native(group) for _, group in data_by_tf[base_tf].groupby('Symbol')]\n", " df_base = pd.concat(df_base_list).reset_index()\n", " df_merged = df_base\n", "\n", " if medium_tf and medium_tf in data_by_tf:\n", " df_medium_ctx = self._calculate_htf_features(data_by_tf[medium_tf], medium_tf, 50, 14)\n", " df_merged = pd.merge_asof(df_merged.sort_values('Timestamp'), df_medium_ctx.sort_values('Timestamp'), on='Timestamp', by='Symbol', direction='backward')\n", "\n", " if high_tf and high_tf in data_by_tf:\n", " df_high_ctx = self._calculate_htf_features(data_by_tf[high_tf], high_tf, 20, 14)\n", " df_merged = pd.merge_asof(df_merged.sort_values('Timestamp'), df_high_ctx.sort_values('Timestamp'), on='Timestamp', by='Symbol', direction='backward')\n", "\n", " df_final = df_merged.set_index('Timestamp').copy()\n", "\n", " if medium_tf:\n", " tf_id = medium_tf.upper()\n", " df_final[f'adx_x_{tf_id}_trend'] = df_final['ADX'] * df_final.get(f'{tf_id}_ctx_Trend', 0)\n", " if high_tf:\n", " tf_id = high_tf.upper()\n", " df_final[f'atr_x_{tf_id}_trend'] = df_final['ATR'] * df_final.get(f'{tf_id}_ctx_Trend', 0)\n", "\n", " logger.info(\" - Generating anomaly detection scores...\")\n", " df_final['anomaly_score'] = self._get_anomaly_scores(df_final, self.ANOMALY_FEATURES)\n", "\n", " feature_cols = [c for c in df_final.columns if c not in ['Open','High','Low','Close','RealVolume','Symbol']]\n", " df_final[feature_cols] = df_final.groupby('Symbol')[feature_cols].shift(1)\n", "\n", " df_final.replace([np.inf,-np.inf],np.nan,inplace=True)\n", " df_final.dropna(inplace=True)\n", "\n", " logger.info(f\" - Merged data and created features. Final dataset shape: {df_final.shape}\")\n", " logger.info(\"[SUCCESS] Feature engineering complete.\");return df_final\n", "\n", " def label_outcomes(self,df:pd.DataFrame,lookahead:int)->pd.DataFrame:\n", " logger.info(\" - Generating trade labels with Regime-Adjusted Barriers...\");\n", " labeled_dfs=[self._label_group(group,lookahead) for _,group in df.groupby('Symbol')];return pd.concat(labeled_dfs)\n", "\n", " def _label_group(self,group:pd.DataFrame,lookahead:int)->pd.DataFrame:\n", " if len(group)=tp_long)[0]; time_to_sl_long=np.where(future_lows<=sl_long)[0]\n", " first_tp_long=time_to_tp_long[0] if len(time_to_tp_long)>0 else np.inf\n", " first_sl_long=time_to_sl_long[0] if len(time_to_sl_long)>0 else np.inf\n", "\n", " tp_short,sl_short=prices[i]-tp_dist,prices[i]+sl_dist\n", " time_to_tp_short=np.where(future_lows<=tp_short)[0]; time_to_sl_short=np.where(future_highs>=sl_short)[0]\n", " first_tp_short=time_to_tp_short[0] if len(time_to_tp_short)>0 else np.inf\n", " first_sl_short=time_to_sl_short[0] if len(time_to_sl_short)>0 else np.inf\n", "\n", " if first_tp_long < first_sl_long: outcomes[i]=1\n", " if first_tp_short < first_sl_short: outcomes[i]=-1\n", "\n", " group['target']=outcomes;return group\n", "\n", " def _label_meta_group(self, group: pd.DataFrame, lookahead: int) -> pd.DataFrame:\n", " if 'primary_model_signal' not in group.columns or len(group) < lookahead + 1:\n", " group['target'] = 0\n", " return group\n", "\n", " is_trending = group['market_regime'] == 1\n", " sl_multiplier = np.where(is_trending, 2.0, 1.5)\n", " tp_multiplier = np.where(is_trending, 4.0, 2.5)\n", " sl_atr_dynamic = group['ATR'] * sl_multiplier\n", " tp_atr_dynamic = group['ATR'] * tp_multiplier\n", "\n", " outcomes = np.zeros(len(group))\n", " prices, lows, highs = group['Close'].values, group['Low'].values, group['High'].values\n", " primary_signals = group['primary_model_signal'].values\n", "\n", " for i in range(len(group) - lookahead):\n", " signal = primary_signals[i]\n", " if signal == 0: continue\n", "\n", " sl_dist, tp_dist = sl_atr_dynamic[i], tp_atr_dynamic[i]\n", " if pd.isna(sl_dist) or sl_dist <= 1e-9: continue\n", "\n", " future_highs, future_lows = highs[i + 1:i + 1 + lookahead], lows[i + 1:i + 1 + lookahead]\n", "\n", " if signal > 0:\n", " tp_long, sl_long = prices[i] + tp_dist, prices[i] - sl_dist\n", " time_to_tp = np.where(future_highs >= tp_long)[0]\n", " time_to_sl = np.where(future_lows <= sl_long)[0]\n", " if len(time_to_tp) > 0 and (len(time_to_sl) == 0 or time_to_tp[0] < time_to_sl[0]):\n", " outcomes[i] = 1\n", "\n", " elif signal < 0:\n", " tp_short, sl_short = prices[i] - tp_dist, prices[i] + sl_dist\n", " time_to_tp = np.where(future_lows <= tp_short)[0]\n", " time_to_sl = np.where(future_highs >= sl_short)[0]\n", " if len(time_to_tp) > 0 and (len(time_to_sl) == 0 or time_to_tp[0] < time_to_sl[0]):\n", " outcomes[i] = 1\n", "\n", " group['target'] = outcomes\n", " return group\n", "\n", " def label_meta_outcomes(self, df: pd.DataFrame, lookahead: int) -> pd.DataFrame:\n", " logger.info(\" - Generating BINARY meta-labels (1=correct, 0=incorrect)...\")\n", " labeled_dfs = [self._label_meta_group(group, lookahead) for _, group in df.groupby('Symbol')]\n", " if not labeled_dfs: return pd.DataFrame()\n", " return pd.concat(labeled_dfs)\n", "\n", "# =============================================================================\n", "# 6. MODELS & TRAINER\n", "# =============================================================================\n", "class GNNModel(torch.nn.Module if GNN_AVAILABLE else object):\n", " def __init__(self, in_channels, hidden_channels, out_channels):\n", " super(GNNModel, self).__init__()\n", " self.conv1 = GCNConv(in_channels, hidden_channels)\n", " self.conv2 = GCNConv(hidden_channels, out_channels)\n", "\n", " def forward(self, data):\n", " x, edge_index = data.x, data.edge_index\n", " x = self.conv1(x, edge_index)\n", " x = F.relu(x)\n", " x = F.dropout(x, p=0.5, training=self.training)\n", " x = self.conv2(x, edge_index)\n", " return x\n", "\n", "class ModelTrainer:\n", " GNN_BASE_FEATURES = ['ATR', 'RSI', 'ADX', 'bollinger_bandwidth', 'stoch_k', 'momentum_10', 'hour', 'day_of_week']\n", " def __init__(self,config:ConfigModel):\n", " self.config=config\n", " self.shap_summary:Optional[pd.DataFrame]=None\n", " self.class_weights:Optional[Dict[int,float]]=None\n", " self.best_threshold=0.5\n", " self.study: Optional[optuna.study.Study] = None\n", " self.is_gnn_model = False\n", " self.is_meta_model = False\n", " self.gnn_model: Optional[GNNModel] = None\n", " self.gnn_scaler = MinMaxScaler()\n", " self.asset_map: Dict[str, int] = {}\n", "\n", " def train(self, df_train: pd.DataFrame, feature_list: List[str], strategy_details: Dict) -> Optional[Tuple[Pipeline, float]]:\n", " logger.info(f\" - Starting model training using strategy: '{strategy_details.get('description', 'N/A')}'\")\n", " self.is_gnn_model = strategy_details.get(\"requires_gnn\", False)\n", " self.is_meta_model = strategy_details.get(\"requires_meta_labeling\", False)\n", "\n", " X = pd.DataFrame()\n", "\n", " if self.is_gnn_model:\n", " if not GNN_AVAILABLE:\n", " logger.error(\" - Skipping GNN model training: PyTorch/PyG libraries not found.\")\n", " return None\n", " logger.info(\" - GNN strategy detected. Generating graph embeddings as features...\")\n", " gnn_embeddings = self._train_gnn(df_train)\n", " if gnn_embeddings.empty:\n", " logger.error(\" - GNN embedding generation failed. Aborting cycle.\")\n", " return None\n", " X = gnn_embeddings\n", " feature_list = list(X.columns)\n", " logger.info(f\" - Feature set replaced by {len(feature_list)} GNN embeddings.\")\n", " y_map={-1:0,0:1,1:2}; y=df_train['target'].map(y_map).astype(int); num_classes = 3\n", " else:\n", " if not feature_list:\n", " logger.error(f\" - Training aborted for strategy '{strategy_details.get('description', 'N/A')}': The 'selected_features' list is empty.\")\n", " return None\n", "\n", " X = df_train[feature_list].copy().fillna(0)\n", " if self.is_meta_model:\n", " logger.info(\" - Meta-Labeling strategy detected. Training secondary filter model.\")\n", " y = df_train['target'].astype(int); num_classes = 2\n", " else:\n", " y_map={-1:0,0:1,1:2}; y=df_train['target'].map(y_map).astype(int); num_classes = 3\n", "\n", " if X.empty:\n", " logger.error(\" - Training data (X) is empty after feature selection. Aborting cycle.\")\n", " return None\n", " if len(y.unique()) < num_classes:\n", " logger.warning(f\" - Skipping cycle: Not enough classes ({len(y.unique())}) for {num_classes}-class model.\")\n", " return None\n", "\n", " self.class_weights=dict(zip(np.unique(y),compute_class_weight(class_weight='balanced',classes=np.unique(y),y=y)))\n", " X_train_val, X_stability, y_train_val, y_stability = train_test_split(X, y, test_size=0.1, shuffle=False)\n", " X_train, X_val, y_train, y_val = train_test_split(X_train_val, y_train_val, test_size=0.2, shuffle=False)\n", "\n", " if X_train.empty or X_val.empty:\n", " logger.error(f\" - Training aborted: Data split resulted in an empty training or validation set. (Train shape: {X_train.shape}, Val shape: {X_val.shape})\")\n", " return None\n", "\n", " self.study=self._optimize_hyperparameters(X_train,y_train,X_val,y_val, X_stability, y_stability, num_classes)\n", " if not self.study or not self.study.best_trials:\n", " logger.error(\" - Training aborted: Hyperparameter optimization failed.\")\n", " return None\n", "\n", " logger.info(f\" - Optimization complete. Best Objective Score: {self.study.best_value:.4f}\")\n", " logger.info(f\" - Best params: {self.study.best_params}\")\n", "\n", " self.best_threshold = self._find_best_threshold(self.study.best_params, X_train, y_train, X_val, y_val, num_classes)\n", " final_pipeline=self._train_final_model(self.study.best_params,X_train_val,y_train_val, feature_list, num_classes)\n", "\n", " if final_pipeline is None:\n", " logger.error(\" - Training aborted: Final model training failed.\")\n", " return None\n", "\n", " logger.info(\" - [SUCCESS] Model training complete.\")\n", " return final_pipeline, self.best_threshold\n", "\n", " def _create_graph_data(self, df: pd.DataFrame) -> Tuple[Optional[Data], Dict[str, int]]:\n", " logger.info(\" - Creating graph structure from asset correlations...\")\n", " pivot_df = df.pivot(columns='Symbol', values='Close').ffill().dropna(how='all', axis=1)\n", " if pivot_df.shape[1] < 2:\n", " logger.warning(\" - Not enough assets to build a correlation graph. Skipping GNN.\")\n", " return None, {}\n", "\n", " corr_matrix = pivot_df.corr()\n", " assets = corr_matrix.index.tolist()\n", " asset_map = {asset: i for i, asset in enumerate(assets)}\n", " edge_list = []\n", " for i in range(len(assets)):\n", " for j in range(i + 1, len(assets)):\n", " if abs(corr_matrix.iloc[i, j]) > 0.3:\n", " edge_list.extend([[asset_map[assets[i]], asset_map[assets[j]]], [asset_map[assets[j]], asset_map[assets[i]]]])\n", "\n", " if not edge_list:\n", " logger.warning(\" - No strong correlations found. Creating a fully connected graph as fallback.\")\n", " edge_list = [[i, j] for i in range(len(assets)) for j in range(len(assets)) if i != j]\n", "\n", " edge_index = torch.tensor(edge_list, dtype=torch.long).t().contiguous()\n", " feature_cols = [f for f in self.GNN_BASE_FEATURES if f in df.columns]\n", " node_features = df.groupby('Symbol')[feature_cols].mean().reindex(assets).fillna(0)\n", " node_features_scaled = pd.DataFrame(self.gnn_scaler.fit_transform(node_features), index=node_features.index)\n", " x = torch.tensor(node_features_scaled.values, dtype=torch.float)\n", "\n", " return Data(x=x, edge_index=edge_index), asset_map\n", "\n", " def _train_gnn(self, df: pd.DataFrame) -> pd.DataFrame:\n", " graph_data, self.asset_map = self._create_graph_data(df)\n", " if graph_data is None: return pd.DataFrame()\n", "\n", " self.gnn_model = GNNModel(in_channels=graph_data.num_node_features, hidden_channels=self.config.GNN_EMBEDDING_DIM * 2, out_channels=self.config.GNN_EMBEDDING_DIM)\n", " optimizer = Adam(self.gnn_model.parameters(), lr=0.01, weight_decay=5e-4)\n", " self.gnn_model.train()\n", "\n", " for epoch in range(self.config.GNN_EPOCHS):\n", " optimizer.zero_grad()\n", " out = self.gnn_model(graph_data)\n", " loss = out.mean()\n", " loss.backward()\n", " optimizer.step()\n", "\n", " self.gnn_model.eval()\n", " with torch.no_grad():\n", " embeddings = self.gnn_model(graph_data).numpy()\n", "\n", " embedding_df = pd.DataFrame(embeddings, index=self.asset_map.keys(), columns=[f\"gnn_{i}\" for i in range(self.config.GNN_EMBEDDING_DIM)])\n", " full_embeddings = df['Symbol'].map(embedding_df.to_dict('index')).apply(pd.Series)\n", " full_embeddings.index = df.index\n", " return full_embeddings\n", "\n", " def _get_gnn_embeddings_for_test(self, df_test: pd.DataFrame) -> pd.DataFrame:\n", " if not self.is_gnn_model or self.gnn_model is None or not self.asset_map: return pd.DataFrame()\n", "\n", " feature_cols = [f for f in self.GNN_BASE_FEATURES if f in df_test.columns]\n", " test_node_features = df_test.groupby('Symbol')[feature_cols].mean()\n", " aligned_features = test_node_features.reindex(self.asset_map.keys()).fillna(0)\n", " test_node_features_scaled = pd.DataFrame(self.gnn_scaler.transform(aligned_features), index=aligned_features.index)\n", " x = torch.tensor(test_node_features_scaled.values, dtype=torch.float)\n", "\n", " graph_data, _ = self._create_graph_data(df_test)\n", " if graph_data is None: return pd.DataFrame()\n", "\n", " graph_data.x = x\n", " self.gnn_model.eval()\n", "\n", " with torch.no_grad():\n", " embeddings = self.gnn_model(graph_data).numpy()\n", "\n", " embedding_df = pd.DataFrame(embeddings, index=self.asset_map.keys(), columns=[f\"gnn_{i}\" for i in range(self.config.GNN_EMBEDDING_DIM)])\n", " full_embeddings = df_test['Symbol'].map(embedding_df.to_dict('index')).apply(pd.Series)\n", " full_embeddings.index = df_test.index\n", " return full_embeddings\n", "\n", " def _find_best_threshold(self, best_params, X_train, y_train, X_val, y_val, num_classes) -> float:\n", " logger.info(\" - Tuning classification threshold for F1 score...\")\n", " objective = 'multi:softprob' if num_classes > 2 else 'binary:logistic'\n", " temp_params = {'objective':objective,'booster':'gbtree','tree_method':'hist',**best_params}\n", " if num_classes > 2: temp_params['num_class'] = num_classes\n", " temp_params.pop('early_stopping_rounds', None)\n", "\n", " temp_pipeline = Pipeline([('scaler', RobustScaler()), ('model', xgb.XGBClassifier(**temp_params))])\n", " fit_params={'model__sample_weight':y_train.map(self.class_weights)}\n", " temp_pipeline.fit(X_train, y_train, **fit_params)\n", " probs = temp_pipeline.predict_proba(X_val)\n", "\n", " best_f1, best_thresh = -1, 0.5\n", " for threshold in np.arange(0.3, 0.7, 0.01):\n", " if num_classes > 2:\n", " max_probs = np.max(probs, axis=1)\n", " preds = np.argmax(probs, axis=1)\n", " preds = np.where(max_probs > threshold, preds, 1) # Default to 'hold' class\n", " else:\n", " preds = (probs[:, 1] > threshold).astype(int)\n", "\n", " f1 = f1_score(y_val, preds, average='macro', zero_division=0)\n", " if f1 > best_f1:\n", " best_f1, best_thresh = f1, threshold\n", "\n", " logger.info(f\" - Best threshold found: {best_thresh:.2f} (F1: {best_f1:.4f})\")\n", " return best_thresh\n", "\n", " def _optimize_hyperparameters(self,X_train,y_train,X_val,y_val, X_stability, y_stability, num_classes)->Optional[optuna.study.Study]:\n", " logger.info(f\" - Starting hyperparameter optimization with risk-adjusted objective ({self.config.OPTUNA_TRIALS} trials)...\")\n", "\n", " def dynamic_progress_callback(study: optuna.study.Study, trial: optuna.trial.FrozenTrial):\n", " n_trials = self.config.OPTUNA_TRIALS\n", " trial_number = trial.number + 1\n", " best_value = study.best_value if study.best_trial else float('nan')\n", " progress_str = f\"> Optuna Optimization: Trial {trial_number}/{n_trials} | Best Score: {best_value:.4f}\"\n", " sys.stdout.write(f\"\\r{progress_str.ljust(80)}\")\n", " sys.stdout.flush()\n", "\n", " objective = 'multi:softprob' if num_classes > 2 else 'binary:logistic'\n", " eval_metric = 'mlogloss' if num_classes > 2 else 'logloss'\n", "\n", " def custom_objective(trial:optuna.Trial):\n", " param={'objective':objective,'eval_metric':eval_metric,'booster':'gbtree','tree_method':'hist','seed':42,\n", " 'n_estimators':trial.suggest_int('n_estimators',200,1000,step=50),\n", " 'max_depth':trial.suggest_int('max_depth',3,8),\n", " 'learning_rate':trial.suggest_float('learning_rate',0.01,0.2,log=True),\n", " 'subsample':trial.suggest_float('subsample',0.6,1.0),\n", " 'colsample_bytree':trial.suggest_float('colsample_bytree',0.6,1.0),\n", " 'gamma':trial.suggest_float('gamma',0,5),\n", " 'reg_lambda':trial.suggest_float('reg_lambda',1e-8,5.0,log=True),\n", " 'alpha':trial.suggest_float('alpha',1e-8,5.0,log=True),\n", " 'early_stopping_rounds':50}\n", "\n", " if num_classes > 2: param['num_class'] = num_classes\n", "\n", " try:\n", " scaler=RobustScaler()\n", " X_train_scaled=scaler.fit_transform(X_train)\n", " X_val_scaled=scaler.transform(X_val)\n", " model=xgb.XGBClassifier(**param)\n", " fit_params={'sample_weight':y_train.map(self.class_weights)}\n", " model.fit(X_train_scaled,y_train,eval_set=[(X_val_scaled,y_val)],verbose=False,**fit_params)\n", " preds=model.predict(X_val_scaled)\n", " f1 = f1_score(y_val,preds,average='macro', zero_division=0)\n", " pnl_map = {0: -1, 1: 0, 2: 1} if num_classes > 2 else {0: -1, 1: 1}\n", " pnl = pd.Series(preds).map(pnl_map)\n", " downside_returns = pnl[pnl < 0]\n", " downside_std = downside_returns.std()\n", " sortino = (pnl.mean() / downside_std) if downside_std > 0 else 0\n", " objective_score = (0.4 * f1) + (0.6 * sortino)\n", "\n", " X_stability_scaled = scaler.transform(X_stability)\n", " stability_preds = model.predict(X_stability_scaled)\n", " stability_pnl = pd.Series(stability_preds).map(pnl_map)\n", " if stability_pnl.sum() < 0: objective_score -= 0.5\n", "\n", " return objective_score\n", " except Exception as e:\n", " sys.stdout.write(\"\\n\")\n", " logger.warning(f\"Trial {trial.number} failed with error: {e}\")\n", " return -2.0\n", "\n", " try:\n", " study=optuna.create_study(direction='maximize')\n", " study.optimize(custom_objective, n_trials=self.config.OPTUNA_TRIALS, timeout=3600, n_jobs=-1, callbacks=[dynamic_progress_callback])\n", " sys.stdout.write(\"\\n\")\n", " return study\n", " except Exception as e:\n", " sys.stdout.write(\"\\n\")\n", " logger.error(f\" - Optuna study failed catastrophically: {e}\",exc_info=True)\n", " return None\n", "\n", " def _train_final_model(self,best_params:Dict,X:pd.DataFrame,y:pd.Series, feature_names: List[str], num_classes: int)->Optional[Pipeline]:\n", " logger.info(\" - Training final model...\")\n", " try:\n", " best_params.pop('early_stopping_rounds', None)\n", " objective = 'multi:softprob' if num_classes > 2 else 'binary:logistic'\n", " final_params={'objective':objective,'booster':'gbtree','tree_method':'hist','seed':42,**best_params}\n", " if num_classes > 2: final_params['num_class'] = num_classes\n", "\n", " final_pipeline=Pipeline([('scaler',RobustScaler()),('model',xgb.XGBClassifier(**final_params))])\n", " fit_params={'model__sample_weight':y.map(self.class_weights)}\n", " final_pipeline.fit(X,y,**fit_params)\n", "\n", " if self.config.CALCULATE_SHAP_VALUES:\n", " self._generate_shap_summary(final_pipeline.named_steps['model'],final_pipeline.named_steps['scaler'].transform(X), feature_names, num_classes)\n", "\n", " return final_pipeline\n", " except Exception as e:\n", " logger.error(f\" - Error during final model training: {e}\",exc_info=True)\n", " return None\n", "\n", " def _generate_shap_summary(self, model: xgb.XGBClassifier, X_scaled: np.ndarray, feature_names: List[str], num_classes: int):\n", " logger.info(\" - Generating SHAP feature importance summary...\")\n", " try:\n", " if len(X_scaled) > 2000:\n", " logger.info(f\" - Subsampling data for SHAP from {len(X_scaled)} to 2000 rows.\")\n", " np.random.seed(42)\n", " sample_indices = np.random.choice(X_scaled.shape[0], 2000, replace=False)\n", " X_sample = X_scaled[sample_indices]\n", " else:\n", " X_sample = X_scaled\n", "\n", " explainer = shap.TreeExplainer(model)\n", " shap_explanation = explainer(X_sample)\n", "\n", " if num_classes > 2:\n", " mean_abs_shap_per_class = shap_explanation.abs.mean(0).values\n", " overall_importance = mean_abs_shap_per_class.mean(axis=1) if mean_abs_shap_per_class.ndim == 2 else mean_abs_shap_per_class\n", " else: # Binary classification\n", " overall_importance = np.abs(shap_explanation.values).mean(axis=0)\n", "\n", " summary = pd.DataFrame(overall_importance, index=feature_names, columns=['SHAP_Importance']).sort_values(by='SHAP_Importance', ascending=False)\n", " self.shap_summary = summary\n", " logger.info(\" - SHAP summary generated successfully.\")\n", " except Exception as e:\n", " logger.error(f\" - Failed to generate SHAP summary: {e}\", exc_info=True)\n", " self.shap_summary = None\n", "\n", "# =============================================================================\n", "# 7. BACKTESTER & 8. PERFORMANCE ANALYZER\n", "# =============================================================================\n", "class Backtester:\n", " def __init__(self,config:ConfigModel):\n", " self.config=config\n", " self.is_meta_model = False\n", "\n", " def run_backtest_chunk(self, df_chunk_in: pd.DataFrame, confidence_threshold: float, initial_equity: float, is_meta_model: bool) -> Tuple[pd.DataFrame, pd.Series, bool, Optional[Dict], Dict]:\n", " if df_chunk_in.empty:\n", " return pd.DataFrame(), pd.Series([initial_equity]), False, None, {}\n", "\n", " df_chunk = df_chunk_in.copy()\n", " self.is_meta_model = is_meta_model\n", " trades, equity, equity_curve, open_positions = [], initial_equity, [initial_equity], {}\n", " chunk_peak_equity = initial_equity\n", " circuit_breaker_tripped = False\n", " breaker_context = None\n", " candles = df_chunk.reset_index().to_dict('records')\n", "\n", " daily_dd_report = {}\n", " current_day = None\n", " day_start_equity = initial_equity\n", " day_peak_equity = initial_equity\n", "\n", " def finalize_day_metrics(day_to_finalize, equity_at_close):\n", " if day_to_finalize is None: return\n", " daily_pnl = equity_at_close - day_start_equity\n", " daily_dd_pct = ((day_peak_equity - equity_at_close) / day_peak_equity) * 100 if day_peak_equity > 0 else 0\n", " daily_dd_report[day_to_finalize.isoformat()] = {'pnl': round(daily_pnl, 2), 'drawdown_pct': round(daily_dd_pct, 2)}\n", "\n", " for candle in candles:\n", " candle_date = candle['Timestamp'].date()\n", " if candle_date != current_day:\n", " finalize_day_metrics(current_day, equity)\n", " current_day, day_start_equity, day_peak_equity = candle_date, equity, equity\n", "\n", " if not circuit_breaker_tripped:\n", " day_peak_equity = max(day_peak_equity, equity)\n", " chunk_peak_equity = max(chunk_peak_equity, equity)\n", " if equity > 0 and chunk_peak_equity > 0 and (chunk_peak_equity - equity) / chunk_peak_equity > self.config.MAX_DD_PER_CYCLE:\n", " logger.warning(f\" - CYCLE CIRCUIT BREAKER TRIPPED! Drawdown exceeded {self.config.MAX_DD_PER_CYCLE:.0%} for this cycle. Closing all positions.\")\n", " circuit_breaker_tripped = True\n", " trade_df = pd.DataFrame(trades)\n", " breaker_context = {\"num_trades_before_trip\": len(trade_df), \"pnl_before_trip\": round(trade_df['PNL'].sum(), 2), \"last_5_trades_pnl\": [round(p, 2) for p in trade_df['PNL'].tail(5).tolist()]} if not trade_df.empty else {}\n", " open_positions.clear()\n", "\n", " if equity <= 0:\n", " logger.critical(\" - ACCOUNT BLOWN!\")\n", " break\n", "\n", " symbol = candle['Symbol']\n", "\n", " if symbol in open_positions:\n", " pos = open_positions[symbol]\n", " pnl, exit_price, exit_reason = 0, None, \"\"\n", "\n", " if self.config.USE_PARTIAL_PROFIT and not pos['partial_profit_taken']:\n", " partial_tp_price = pos['entry_price'] + (pos['sl_dist'] * self.config.PARTIAL_PROFIT_TRIGGER_R * pos['direction'])\n", " if (pos['direction'] == 1 and candle['High'] >= partial_tp_price) or \\\n", " (pos['direction'] == -1 and candle['Low'] <= partial_tp_price):\n", "\n", " partial_pnl = pos['risk_amt'] * self.config.PARTIAL_PROFIT_TRIGGER_R * self.config.PARTIAL_PROFIT_TAKE_PCT\n", " equity += partial_pnl\n", " day_peak_equity = max(day_peak_equity, equity)\n", " equity_curve.append(equity)\n", "\n", " trades.append({\n", " 'ExecTime': candle['Timestamp'], 'Symbol': symbol, 'PNL': partial_pnl,\n", " 'Equity': equity, 'Confidence': pos['confidence'], 'Direction': pos['direction'],\n", " 'ExitReason': f\"Partial TP ({self.config.PARTIAL_PROFIT_TAKE_PCT:.0%})\"\n", " })\n", "\n", " pos['risk_amt'] *= (1 - self.config.PARTIAL_PROFIT_TAKE_PCT)\n", " pos['sl'] = pos['entry_price']\n", " pos['partial_profit_taken'] = True\n", " if LOG_PARTIAL_PROFITS:\n", " logger.info(f\" - Partial profit taken for {symbol}. Moved SL to breakeven.\")\n", "\n", " if pos['direction'] == 1:\n", " if candle['Low'] <= pos['sl']: pnl, exit_price, exit_reason = -pos['risk_amt'], pos['sl'], \"Stop Loss\"\n", " elif candle['High'] >= pos['tp']: pnl, exit_price, exit_reason = pos['risk_amt'] * pos['rr'], pos['tp'], \"Take Profit\"\n", " elif pos['direction'] == -1:\n", " if candle['High'] >= pos['sl']: pnl, exit_price, exit_reason = -pos['risk_amt'], pos['sl'], \"Stop Loss\"\n", " elif candle['Low'] <= pos['tp']: pnl, exit_price, exit_reason = pos['risk_amt'] * pos['rr'], pos['tp'], \"Take Profit\"\n", "\n", " if exit_price:\n", " equity += pnl\n", " day_peak_equity = max(day_peak_equity, equity)\n", " equity_curve.append(equity)\n", "\n", " trades.append({\n", " 'ExecTime': candle['Timestamp'], 'Symbol': symbol, 'PNL': pnl,\n", " 'Equity': equity, 'Confidence': pos['confidence'], 'Direction': pos['direction'],\n", " 'ExitReason': exit_reason\n", " })\n", " del open_positions[symbol]\n", " if equity <= 0: continue\n", "\n", " if not circuit_breaker_tripped and symbol not in open_positions and len(open_positions) < self.config.MAX_CONCURRENT_TRADES:\n", "\n", " if candle.get('anomaly_score') == -1:\n", " if LOG_ANOMALY_SKIPS and random.random() < 0.1:\n", " logger.info(f\" - Skipping trade check for {symbol} due to anomaly detection at {candle['Timestamp']}\")\n", " continue\n", "\n", " vol_idx = candle.get('market_volatility_index', 0.5)\n", " if not (self.config.MIN_VOLATILITY_RANK <= vol_idx <= self.config.MAX_VOLATILITY_RANK):\n", " continue\n", "\n", " direction, confidence = 0, 0\n", " if self.is_meta_model:\n", " prob_take_trade = candle.get('prob_1', 0)\n", " primary_signal = candle.get('primary_model_signal', 0)\n", " if prob_take_trade > confidence_threshold and primary_signal != 0:\n", " direction = int(np.sign(primary_signal))\n", " confidence = prob_take_trade\n", " else:\n", " if 'prob_short' in candle:\n", " probs=np.array([candle['prob_short'],candle['prob_hold'],candle['prob_long']])\n", " max_confidence=np.max(probs)\n", " if max_confidence >= confidence_threshold:\n", " pred_class=np.argmax(probs)\n", " direction=1 if pred_class==2 else -1 if pred_class==0 else 0\n", " confidence = max_confidence\n", "\n", " if direction != 0:\n", " atr=candle.get('ATR',0)\n", " if pd.isna(atr) or atr<=1e-9: continue\n", "\n", " if confidence>=self.config.CONFIDENCE_TIERS['ultra_high']['min']: tier_name='ultra_high'\n", " elif confidence>=self.config.CONFIDENCE_TIERS['high']['min']: tier_name='high'\n", " else: tier_name='standard'\n", "\n", " tier=self.config.CONFIDENCE_TIERS[tier_name]\n", " base_risk_amt = equity * self.config.BASE_RISK_PER_TRADE_PCT * tier['risk_mult']\n", " risk_amt = min(base_risk_amt, self.config.RISK_CAP_PER_TRADE_USD)\n", " sl_dist=(atr*1.5)+(atr*self.config.SPREAD_PCTG_OF_ATR)+(atr*self.config.SLIPPAGE_PCTG_OF_ATR)\n", " tp_dist=(atr*1.5*tier['rr'])\n", " if tp_dist<=0 or sl_dist<=0: continue\n", "\n", " entry_price=candle['Close']\n", " sl_price,tp_price=entry_price-sl_dist*direction,entry_price+tp_dist*direction\n", "\n", " open_positions[symbol]={\n", " 'direction':direction, 'entry_price': entry_price, 'sl':sl_price,'tp':tp_price,\n", " 'risk_amt':risk_amt, 'rr':tier['rr'], 'confidence':confidence,\n", " 'sl_dist': sl_dist, 'partial_profit_taken': False\n", " }\n", "\n", " day_peak_equity = max(day_peak_equity, equity)\n", "\n", " finalize_day_metrics(current_day, equity)\n", " return pd.DataFrame(trades), pd.Series(equity_curve), circuit_breaker_tripped, breaker_context, daily_dd_report\n", "\n", "class PerformanceAnalyzer:\n", " def __init__(self,config:ConfigModel):\n", " self.config=config\n", "\n", " def generate_full_report(self,trades_df:Optional[pd.DataFrame],equity_curve:Optional[pd.Series],cycle_metrics:List[Dict],aggregated_shap:Optional[pd.DataFrame]=None, framework_memory:Optional[Dict]=None, aggregated_daily_dd:Optional[List[Dict]]=None) -> Dict[str, Any]:\n", " logger.info(\"-> Stage 4: Generating Final Performance Report...\")\n", " if equity_curve is not None and len(equity_curve) > 1: self.plot_equity_curve(equity_curve)\n", " if aggregated_shap is not None: self.plot_shap_summary(aggregated_shap)\n", "\n", " metrics = self._calculate_metrics(trades_df, equity_curve) if trades_df is not None and not trades_df.empty else {}\n", " self.generate_text_report(metrics, cycle_metrics, aggregated_shap, framework_memory, aggregated_daily_dd)\n", "\n", " logger.info(f\"[SUCCESS] Final report generated and saved to: {self.config.REPORT_SAVE_PATH}\")\n", " return metrics\n", "\n", " def plot_equity_curve(self,equity_curve:pd.Series):\n", " plt.style.use('seaborn-v0_8-darkgrid')\n", " plt.figure(figsize=(16,8))\n", " plt.plot(equity_curve.values,color='dodgerblue',linewidth=2)\n", " plt.title(f\"{self.config.nickname or self.config.REPORT_LABEL} - Walk-Forward Equity Curve\",fontsize=16,weight='bold')\n", " plt.xlabel(\"Trade Event Number (including partial closes)\",fontsize=12)\n", " plt.ylabel(\"Equity ($)\",fontsize=12)\n", " plt.grid(True,which='both',linestyle=':')\n", " try:\n", " plt.savefig(self.config.PLOT_SAVE_PATH)\n", " plt.close()\n", " logger.info(f\" - Equity curve plot saved to: {self.config.PLOT_SAVE_PATH}\")\n", " except Exception as e:\n", " logger.error(f\" - Failed to save equity curve plot: {e}\")\n", "\n", " def plot_shap_summary(self,shap_summary:pd.DataFrame):\n", " plt.style.use('seaborn-v0_8-darkgrid')\n", " plt.figure(figsize=(12,10))\n", " shap_summary.head(20).sort_values(by='SHAP_Importance').plot(kind='barh',legend=False,color='mediumseagreen')\n", " title_str = f\"{self.config.nickname or self.config.REPORT_LABEL} ({self.config.strategy_name}) - Aggregated Feature Importance\"\n", " plt.title(title_str,fontsize=16,weight='bold')\n", " plt.xlabel(\"Mean Absolute SHAP Value\",fontsize=12)\n", " plt.ylabel(\"Feature\",fontsize=12)\n", " plt.tight_layout()\n", " try:\n", " plt.savefig(self.config.SHAP_PLOT_PATH)\n", " plt.close()\n", " logger.info(f\" - SHAP summary plot saved to: {self.config.SHAP_PLOT_PATH}\")\n", " except Exception as e:\n", " logger.error(f\" - Failed to save SHAP plot: {e}\")\n", "\n", " def _calculate_metrics(self,trades_df:pd.DataFrame,equity_curve:pd.Series)->Dict[str,Any]:\n", " m={}\n", " m['initial_capital']=self.config.INITIAL_CAPITAL\n", " m['ending_capital']=equity_curve.iloc[-1]\n", " m['total_net_profit']=m['ending_capital']-m['initial_capital']\n", " m['net_profit_pct']=(m['total_net_profit']/m['initial_capital']) if m['initial_capital']>0 else 0\n", "\n", " returns=trades_df['PNL']/m['initial_capital']\n", " wins=trades_df[trades_df['PNL']>0]\n", " losses=trades_df[trades_df['PNL']<0]\n", " m['gross_profit']=wins['PNL'].sum()\n", " m['gross_loss']=abs(losses['PNL'].sum())\n", " m['profit_factor']=m['gross_profit']/m['gross_loss'] if m['gross_loss']>0 else np.inf\n", "\n", " m['total_trade_events']=len(trades_df)\n", " final_exits_df = trades_df[trades_df['ExitReason'].isin([\"Stop Loss\", \"Take Profit\"])]\n", " m['total_trades'] = len(final_exits_df)\n", "\n", " m['winning_trades']=len(final_exits_df[final_exits_df['PNL'] > 0])\n", " m['losing_trades']=len(final_exits_df[final_exits_df['PNL'] < 0])\n", " m['win_rate']=m['winning_trades']/m['total_trades'] if m['total_trades']>0 else 0\n", "\n", " m['avg_win_amount']=wins['PNL'].mean() if len(wins)>0 else 0\n", " m['avg_loss_amount']=abs(losses['PNL'].mean()) if len(losses)>0 else 0\n", "\n", " avg_full_win = final_exits_df[final_exits_df['PNL'] > 0]['PNL'].mean() if len(final_exits_df[final_exits_df['PNL'] > 0]) > 0 else 0\n", " avg_full_loss = abs(final_exits_df[final_exits_df['PNL'] < 0]['PNL'].mean()) if len(final_exits_df[final_exits_df['PNL'] < 0]) > 0 else 0\n", " m['payoff_ratio']=avg_full_win/avg_full_loss if avg_full_loss > 0 else np.inf\n", " m['expected_payoff']=(m['win_rate']*avg_full_win)-((1-m['win_rate'])*avg_full_loss) if m['total_trades']>0 else 0\n", "\n", " running_max=equity_curve.cummax()\n", " drawdown_abs=running_max-equity_curve\n", " m['max_drawdown_abs']=drawdown_abs.max() if not drawdown_abs.empty else 0\n", " m['max_drawdown_pct']=((drawdown_abs/running_max).replace([np.inf,-np.inf],0).max())*100\n", "\n", " exec_times=pd.to_datetime(trades_df['ExecTime']).dt.tz_localize(None)\n", " years=((exec_times.max()-exec_times.min()).days/365.25) if not trades_df.empty else 1\n", " years = max(years, 1/365.25)\n", " m['cagr']=(((m['ending_capital']/m['initial_capital'])**(1/years))-1) if years>0 and m['initial_capital']>0 else 0\n", "\n", " pnl_std=returns.std()\n", " m['sharpe_ratio']=(returns.mean()/pnl_std)*np.sqrt(252*24*4) if pnl_std>0 else 0\n", " downside_returns=returns[returns<0]\n", " downside_std=downside_returns.std()\n", " m['sortino_ratio']=(returns.mean()/downside_std)*np.sqrt(252*24*4) if downside_std>0 else np.inf\n", " m['calmar_ratio']=m['cagr']/(m['max_drawdown_pct']/100) if m['max_drawdown_pct']>0 else np.inf\n", " m['mar_ratio']=m['calmar_ratio']\n", " m['recovery_factor']=m['total_net_profit']/m['max_drawdown_abs'] if m['max_drawdown_abs']>0 else np.inf\n", "\n", " pnl_series = final_exits_df['PNL']\n", " win_streaks = (pnl_series > 0).astype(int).groupby((pnl_series <= 0).cumsum()).cumsum()\n", " loss_streaks = (pnl_series < 0).astype(int).groupby((pnl_series >= 0).cumsum()).cumsum()\n", " m['longest_win_streak'] = win_streaks.max() if not win_streaks.empty else 0\n", " m['longest_loss_streak'] = loss_streaks.max() if not loss_streaks.empty else 0\n", " return m\n", "\n", " def _get_comparison_block(self, metrics: Dict, memory: Dict, ledger: Dict, width: int) -> str:\n", " champion = memory.get('champion_config')\n", " historical_runs = memory.get('historical_runs', [])\n", " previous_run = historical_runs[-1] if historical_runs else None\n", "\n", " def get_data(source: Optional[Dict], key: str, is_percent: bool = False) -> str:\n", " if not source: return \"N/A\"\n", " val = source.get(key) if isinstance(source, dict) and key in source else source.get(\"final_metrics\", {}).get(key) if isinstance(source, dict) else None\n", " if val is None or not isinstance(val, (int, float)): return \"N/A\"\n", " return f\"{val:.2f}%\" if is_percent else f\"{val:.2f}\"\n", "\n", " def get_info(source: Optional[Union[Dict, ConfigModel]], key: str) -> str:\n", " if not source: return \"N/A\"\n", " if hasattr(source, key):\n", " return str(getattr(source, key, 'N/A'))\n", " elif isinstance(source, dict):\n", " return str(source.get(key, 'N/A'))\n", " return \"N/A\"\n", "\n", " def get_nickname(source: Optional[Union[Dict, ConfigModel]]) -> str:\n", " if not source: return \"N/A\"\n", " version_key = 'REPORT_LABEL' if hasattr(source, 'REPORT_LABEL') else 'script_version'\n", " version = get_info(source, version_key)\n", " return ledger.get(version, \"N/A\")\n", "\n", " c_nick, p_nick, champ_nick = get_nickname(self.config), get_nickname(previous_run), get_nickname(champion)\n", " c_strat, p_strat, champ_strat = get_info(self.config, 'strategy_name'), get_info(previous_run, 'strategy_name'), get_info(champion, 'strategy_name')\n", " c_mar, p_mar, champ_mar = get_data(metrics, 'mar_ratio'), get_data(previous_run, 'mar_ratio'), get_data(champion, 'mar_ratio')\n", " c_mdd, p_mdd, champ_mdd = get_data(metrics, 'max_drawdown_pct', True), get_data(previous_run, 'max_drawdown_pct', True), get_data(champion, 'max_drawdown_pct', True)\n", " c_pf, p_pf, champ_pf = get_data(metrics, 'profit_factor'), get_data(previous_run, 'profit_factor'), get_data(champion, 'profit_factor')\n", "\n", " col_w = (width - 5) // 4\n", " header = f\"| {'Metric'.ljust(col_w-1)}|{'Current Run'.center(col_w)}|{'Previous Run'.center(col_w)}|{'All-Time Champion'.center(col_w)}|\"\n", " sep = f\"+{'-'*(col_w)}+{'-'*(col_w)}+{'-'*(col_w)}+{'-'*(col_w)}+\"\n", " rows = [\n", " f\"| {'Run Nickname'.ljust(col_w-1)}|{c_nick.center(col_w)}|{p_nick.center(col_w)}|{champ_nick.center(col_w)}|\",\n", " f\"| {'Strategy'.ljust(col_w-1)}|{c_strat.center(col_w)}|{p_strat.center(col_w)}|{champ_strat.center(col_w)}|\",\n", " f\"| {'MAR Ratio'.ljust(col_w-1)}|{c_mar.center(col_w)}|{p_mar.center(col_w)}|{champ_mar.center(col_w)}|\",\n", " f\"| {'Max Drawdown'.ljust(col_w-1)}|{c_mdd.center(col_w)}|{p_mdd.center(col_w)}|{champ_mdd.center(col_w)}|\",\n", " f\"| {'Profit Factor'.ljust(col_w-1)}|{c_pf.center(col_w)}|{p_pf.center(col_w)}|{champ_pf.center(col_w)}|\"\n", " ]\n", " return \"\\n\".join([header, sep] + rows)\n", "\n", " def generate_text_report(self, m: Dict[str, Any], cycle_metrics: List[Dict], aggregated_shap: Optional[pd.DataFrame] = None, framework_memory: Optional[Dict] = None, aggregated_daily_dd: Optional[List[Dict]] = None):\n", " WIDTH = 90\n", " def _box_top(w): return f\"+{'-' * (w-2)}+\"\n", " def _box_mid(w): return f\"+{'-' * (w-2)}+\"\n", " def _box_bot(w): return f\"+{'-' * (w-2)}+\"\n", " def _box_line(text, w):\n", " padding = w - 4 - len(text)\n", " return f\"| {text}{' ' * padding} |\" if padding >= 0 else f\"| {text[:w-5]}... |\"\n", " def _box_title(title, w): return f\"| {title.center(w-4)} |\"\n", " def _box_text_kv(key, val, w):\n", " val_str = str(val)\n", " key_len = len(key)\n", " padding = w - 4 - key_len - len(val_str)\n", " return f\"| {key}{' ' * padding}{val_str} |\"\n", "\n", " ledger = {};\n", " if self.config.NICKNAME_LEDGER_PATH and os.path.exists(self.config.NICKNAME_LEDGER_PATH):\n", " try:\n", " with open(self.config.NICKNAME_LEDGER_PATH, 'r') as f: ledger = json.load(f)\n", " except (json.JSONDecodeError, IOError): logger.warning(\"Could not load nickname ledger for reporting.\")\n", "\n", " report = [_box_top(WIDTH)]\n", " report.append(_box_title('ADAPTIVE WALK-FORWARD PERFORMANCE REPORT', WIDTH))\n", " report.append(_box_mid(WIDTH))\n", " report.append(_box_line(f\"Nickname: {self.config.nickname or 'N/A'} ({self.config.strategy_name})\", WIDTH))\n", " report.append(_box_line(f\"Version: {self.config.REPORT_LABEL}\", WIDTH))\n", " report.append(_box_line(f\"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\", WIDTH))\n", "\n", " if self.config.analysis_notes:\n", " report.append(_box_line(f\"AI Notes: {self.config.analysis_notes}\", WIDTH))\n", "\n", " if framework_memory:\n", " report.append(_box_mid(WIDTH))\n", " report.append(_box_title('I. PERFORMANCE vs. HISTORY', WIDTH))\n", " report.append(_box_mid(WIDTH))\n", " report.append(self._get_comparison_block(m, framework_memory, ledger, WIDTH))\n", "\n", " sections = {\n", " \"II. EXECUTIVE SUMMARY\": [\n", " (f\"Initial Capital:\", f\"${m.get('initial_capital', 0):>15,.2f}\"),\n", " (f\"Ending Capital:\", f\"${m.get('ending_capital', 0):>15,.2f}\"),\n", " (f\"Total Net Profit:\", f\"${m.get('total_net_profit', 0):>15,.2f} ({m.get('net_profit_pct', 0):.2%})\"),\n", " (f\"Profit Factor:\", f\"{m.get('profit_factor', 0):>15.2f}\"),\n", " (f\"Win Rate (Full Trades):\", f\"{m.get('win_rate', 0):>15.2%}\"),\n", " (f\"Expected Payoff:\", f\"${m.get('expected_payoff', 0):>15.2f}\")\n", " ],\n", " \"III. CORE PERFORMANCE METRICS\": [\n", " (f\"Annual Return (CAGR):\", f\"{m.get('cagr', 0):>15.2%}\"),\n", " (f\"Sharpe Ratio (annual):\", f\"{m.get('sharpe_ratio', 0):>15.2f}\"),\n", " (f\"Sortino Ratio (annual):\", f\"{m.get('sortino_ratio', 0):>15.2f}\"),\n", " (f\"Calmar Ratio / MAR:\", f\"{m.get('mar_ratio', 0):>15.2f}\")\n", " ],\n", " \"IV. RISK & DRAWDOWN ANALYSIS\": [\n", " (f\"Max Drawdown (Cycle):\", f\"{m.get('max_drawdown_pct', 0):>15.2f}% (${m.get('max_drawdown_abs', 0):,.2f})\"),\n", " (f\"Recovery Factor:\", f\"{m.get('recovery_factor', 0):>15.2f}\"),\n", " (f\"Longest Losing Streak:\", f\"{m.get('longest_loss_streak', 0):>15} trades\")\n", " ],\n", " \"V. TRADE-LEVEL STATISTICS\": [\n", " (f\"Total Unique Trades:\", f\"{m.get('total_trades', 0):>15}\"),\n", " (f\"Total Trade Events (incl. partials):\", f\"{m.get('total_trade_events', 0):>15}\"),\n", " (f\"Average Win Event:\", f\"${m.get('avg_win_amount', 0):>15,.2f}\"),\n", " (f\"Average Loss Event:\", f\"${m.get('avg_loss_amount', 0):>15,.2f}\"),\n", " (f\"Payoff Ratio (Full Trades):\", f\"{m.get('payoff_ratio', 0):>15.2f}\")\n", " ]\n", " }\n", " for title, data in sections.items():\n", " if not m: continue\n", " report.append(_box_mid(WIDTH))\n", " report.append(_box_title(title, WIDTH))\n", " report.append(_box_mid(WIDTH))\n", " for key, val in data: report.append(_box_text_kv(key, val, WIDTH))\n", "\n", " report.append(_box_mid(WIDTH))\n", " report.append(_box_title('VI. WALK-FORWARD CYCLE BREAKDOWN', WIDTH))\n", " report.append(_box_mid(WIDTH))\n", " cycle_df_str = pd.DataFrame(cycle_metrics).to_string(index=False) if not pd.DataFrame(cycle_metrics).empty else \"No trades were executed.\"\n", " for line in cycle_df_str.split('\\n'): report.append(_box_line(line, WIDTH))\n", "\n", " report.append(_box_mid(WIDTH))\n", " report.append(_box_title('VII. MODEL FEATURE IMPORTANCE (TOP 15)', WIDTH))\n", " report.append(_box_mid(WIDTH))\n", " shap_str = aggregated_shap.head(15).to_string() if aggregated_shap is not None else \"SHAP summary was not generated.\"\n", " for line in shap_str.split('\\n'): report.append(_box_line(line, WIDTH))\n", "\n", " if aggregated_daily_dd:\n", " report.append(_box_mid(WIDTH))\n", " report.append(_box_title('VIII. HIGH DAILY DRAWDOWN EVENTS (>15%)', WIDTH))\n", " report.append(_box_mid(WIDTH))\n", " high_dd_events = []\n", " for cycle_idx, cycle_dd_report in enumerate(aggregated_daily_dd):\n", " for day, data in cycle_dd_report.items():\n", " if data['drawdown_pct'] > 15.0:\n", " high_dd_events.append(f\"Cycle {cycle_idx+1} | {day} | DD: {data['drawdown_pct']:.2f}% | PNL: ${data['pnl']:,.2f}\")\n", "\n", " if high_dd_events:\n", " for event in high_dd_events:\n", " report.append(_box_line(event, WIDTH))\n", " else:\n", " report.append(_box_line(\"No days with drawdown greater than 15% were recorded.\", WIDTH))\n", "\n", " report.append(_box_bot(WIDTH))\n", " final_report = \"\\n\".join(report)\n", " logger.info(\"\\n\" + final_report)\n", " try:\n", " with open(self.config.REPORT_SAVE_PATH,'w',encoding='utf-8') as f: f.write(final_report)\n", " except IOError as e: logger.error(f\" - Failed to save text report: {e}\",exc_info=True)\n", "\n", "# =============================================================================\n", "# 9. FRAMEWORK ORCHESTRATION & MEMORY\n", "# =============================================================================\n", "def _sanitize_ai_suggestions(suggestions: Dict[str, Any]) -> Dict[str, Any]:\n", " \"\"\"Validates and sanitizes critical numeric parameters from the AI.\"\"\"\n", " sanitized = suggestions.copy()\n", " bounds = {\n", " 'MAX_DD_PER_CYCLE': (0.05, 0.99), 'MAX_CONCURRENT_TRADES': (1, 20),\n", " 'PARTIAL_PROFIT_TRIGGER_R': (0.1, 10.0), 'PARTIAL_PROFIT_TAKE_PCT': (0.1, 0.9),\n", " 'OPTUNA_TRIALS': (10, 200), 'LOOKAHEAD_CANDLES': (10, 500)\n", " }\n", " integer_keys = ['MAX_CONCURRENT_TRADES', 'OPTUNA_TRIALS', 'LOOKAHEAD_CANDLES']\n", "\n", " for key, (lower, upper) in bounds.items():\n", " if key in sanitized and isinstance(sanitized.get(key), (int, float)):\n", " original_value = sanitized[key]\n", " clamped_value = max(lower, min(original_value, upper))\n", " if key in integer_keys: clamped_value = int(round(clamped_value))\n", " if original_value != clamped_value:\n", " logger.warning(f\" - Sanitizing AI suggestion for '{key}': Clamped value from {original_value} to {clamped_value} to meet model constraints.\")\n", " sanitized[key] = clamped_value\n", " return sanitized\n", "\n", "def _sanitize_frequency_string(freq_str: Any, default: str = '90D') -> str:\n", " \"\"\"More robustly sanitizes a string to be a valid pandas frequency.\"\"\"\n", " if isinstance(freq_str, int):\n", " sanitized_freq = f\"{freq_str}D\"\n", " logger.warning(f\"AI provided a unit-less number for frequency. Interpreting '{freq_str}' as '{sanitized_freq}'.\")\n", " return sanitized_freq\n", "\n", " if not isinstance(freq_str, str): freq_str = str(freq_str)\n", " if freq_str.isdigit():\n", " sanitized_freq = f\"{freq_str}D\"\n", " logger.warning(f\"AI provided a unit-less string for frequency. Interpreting '{freq_str}' as '{sanitized_freq}'.\")\n", " return sanitized_freq\n", "\n", " try:\n", " pd.tseries.frequencies.to_offset(freq_str)\n", " logger.info(f\"Using valid frequency alias from AI: '{freq_str}'\")\n", " return freq_str\n", " except ValueError:\n", " match = re.search(r'(\\d+)\\s*([A-Za-z]+)', freq_str)\n", " if match:\n", " num, unit_text = match.groups()\n", " unit_map = {'day': 'D', 'days': 'D', 'week': 'W', 'weeks': 'W', 'month': 'M', 'months': 'M'}\n", " unit = unit_map.get(unit_text.lower())\n", " if unit:\n", " sanitized_freq = f\"{num}{unit}\"\n", " logger.warning(f\"Sanitizing AI-provided frequency '{freq_str}' to '{sanitized_freq}'.\")\n", " return sanitized_freq\n", "\n", " logger.error(f\"Could not parse a valid frequency from '{freq_str}'. Falling back to default '{default}'.\")\n", " return default\n", "\n", "def load_memory(champion_path: str, history_path: str) -> Dict:\n", " champion_config = None\n", " if os.path.exists(champion_path):\n", " try:\n", " with open(champion_path, 'r') as f: champion_config = json.load(f)\n", " except (json.JSONDecodeError, IOError) as e: logger.error(f\"Could not read or parse champion file at {champion_path}: {e}\")\n", " historical_runs = []\n", " if os.path.exists(history_path):\n", " try:\n", " with open(history_path, 'r') as f:\n", " for i, line in enumerate(f):\n", " if not line.strip(): continue\n", " try: historical_runs.append(json.loads(line))\n", " except json.JSONDecodeError: logger.warning(f\"Skipping malformed line {i+1} in history file: {history_path}\")\n", " except IOError as e: logger.error(f\"Could not read history file at {history_path}: {e}\")\n", " return {\"champion_config\": champion_config, \"historical_runs\": historical_runs}\n", "\n", "def _recursive_sanitize(data: Any) -> Any:\n", " \"\"\"Recursively traverses a dict/list to convert non-JSON-serializable types.\"\"\"\n", " if isinstance(data, dict):\n", " return {key: _recursive_sanitize(value) for key, value in data.items()}\n", " if isinstance(data, list):\n", " return [_recursive_sanitize(item) for item in data]\n", " if isinstance(data, (np.int64, np.int32)): return int(data)\n", " if isinstance(data, (np.float64, np.float32)):\n", " if np.isnan(data) or np.isinf(data): return None\n", " return float(data)\n", " if isinstance(data, (pd.Timestamp, datetime, date)): return data.isoformat()\n", " if isinstance(data, pathlib.Path): return str(data)\n", " return data\n", "\n", "def save_run_to_memory(config: ConfigModel, new_run_summary: Dict, current_memory: Dict) -> Optional[Dict]:\n", " try:\n", " sanitized_summary = _recursive_sanitize(new_run_summary)\n", " with open(config.HISTORY_FILE_PATH, 'a') as f: f.write(json.dumps(sanitized_summary) + '\\n')\n", " logger.info(f\"-> Run summary appended to history file: {config.HISTORY_FILE_PATH}\")\n", " except IOError as e: logger.error(f\"Could not write to history file: {e}\")\n", "\n", " current_champion = current_memory.get(\"champion_config\")\n", " new_mar = new_run_summary.get(\"final_metrics\", {}).get(\"mar_ratio\", 0)\n", "\n", " is_new_champion = (current_champion is None or\n", " (new_mar is not None and new_mar > current_champion.get(\"final_metrics\", {}).get(\"mar_ratio\", -np.inf)))\n", "\n", " if is_new_champion:\n", " champion_to_save = new_run_summary\n", " champion_mar = current_champion.get(\"final_metrics\", {}).get(\"mar_ratio\", -np.inf) if current_champion else -np.inf\n", " logger.info(f\"NEW CHAMPION! Current run's MAR Ratio ({new_mar:.2f}) beats previous champion's ({champion_mar:.2f}).\")\n", " else:\n", " champion_to_save = current_champion\n", " champ_mar_val = current_champion.get(\"final_metrics\", {}).get(\"mar_ratio\", 0) if current_champion else -np.inf\n", " logger.info(f\"Current run's MAR Ratio ({new_mar:.2f}) did not beat champion's ({champ_mar_val:.2f}).\")\n", "\n", " try:\n", " if champion_to_save:\n", " sanitized_champion = _recursive_sanitize(champion_to_save)\n", " with open(config.CHAMPION_FILE_PATH, 'w') as f: json.dump(sanitized_champion, f, indent=4)\n", " logger.info(f\"-> Champion file updated: {config.CHAMPION_FILE_PATH}\")\n", " except (IOError, TypeError) as e: logger.error(f\"Could not write to champion file: {e}\")\n", "\n", " return champion_to_save\n", "\n", "def initialize_playbook(base_path: str) -> Dict:\n", " results_dir = os.path.join(base_path, \"Results\"); os.makedirs(results_dir, exist_ok=True)\n", " playbook_path = os.path.join(results_dir, \"strategy_playbook.json\")\n", " DEFAULT_PLAYBOOK = {\n", " \"FilteredBreakout\": {\n", " \"description\": \"[BREAKOUT] A hybrid that trades high-volatility breakouts (ATR, Bollinger) but only in the direction of the long-term daily trend.\",\n", " \"features\": [\"ATR\", \"bollinger_bandwidth\", \"DAILY_ctx_Trend\", \"ADX\", \"hour\", \"anomaly_score\"],\n", " \"lookahead_range\": [60, 120], \"dd_range\": [0.2, 0.35]\n", " },\n", " \"RangeBound\": {\n", " \"description\": \"[RANGING] Trades reversals in a sideways channel, filtered by low ADX.\",\n", " \"features\": [\"ADX\", \"RSI\", \"stoch_k\", \"stoch_d\", \"bollinger_bandwidth\", \"hour\", \"is_doji\"],\n", " \"lookahead_range\": [20, 60], \"dd_range\": [0.1, 0.2]\n", " },\n", " \"TrendPullback\": {\n", " \"description\": \"[TRENDING] Enters on pullbacks (RSI) in the direction of an established long-term trend.\",\n", " \"features\": [\"DAILY_ctx_Trend\", \"ADX\", \"RSI\", \"H1_ctx_Trend\", \"momentum_10\", \"hour\", \"close_fracdiff\"],\n", " \"lookahead_range\": [50, 150], \"dd_range\": [0.15, 0.3]\n", " },\n", " \"ConfluenceTrend\": {\n", " \"description\": \"[TRENDING] A more conservative trend strategy that requires confluence from multiple indicators.\",\n", " \"features\": [\"DAILY_ctx_Trend\", \"H1_ctx_Trend\", \"RSI\", \"stoch_k\", \"ADX\", \"momentum_10\"],\n", " \"lookahead_range\": [60, 160], \"dd_range\": [0.15, 0.3]\n", " },\n", " \"MeanReversionOscillator\": {\n", " \"description\": \"[RANGING] A pure mean-reversion strategy using oscillators for entry signals in low-volatility environments.\",\n", " \"features\": [\"RSI\", \"stoch_k\", \"ADX\", \"market_volatility_index\", \"close_fracdiff\", \"hour\", \"day_of_week\"],\n", " \"lookahead_range\": [20, 60], \"dd_range\": [0.15, 0.25]\n", " },\n", " \"GNN_Market_Structure\": {\n", " \"description\": \"[SPECIALIZED] Uses a GNN to model inter-asset correlations for predictive features.\",\n", " \"features\": [], \"lookahead_range\": [80, 150], \"dd_range\": [0.15, 0.3], \"requires_gnn\": True\n", " },\n", " \"Meta_Labeling_Filter\": {\n", " \"description\": \"[SPECIALIZED] Uses a secondary ML filter to improve a simple primary model's signal quality.\",\n", " \"features\": [\"ADX\", \"RSI\", \"ATR\", \"bollinger_bandwidth\", \"H1_ctx_Trend\", \"DAILY_ctx_Trend\", \"momentum_20\"],\n", " \"lookahead_range\": [50, 100], \"dd_range\": [0.1, 0.25], \"requires_meta_labeling\": True\n", " }\n", " }\n", "\n", " if not os.path.exists(playbook_path):\n", " logger.warning(f\"'strategy_playbook.json' not found. Seeding a new one with default strategies at: {playbook_path}\")\n", " try:\n", " with open(playbook_path, 'w') as f: json.dump(DEFAULT_PLAYBOOK, f, indent=4)\n", " return DEFAULT_PLAYBOOK\n", " except IOError as e: logger.error(f\"Failed to create playbook file: {e}. Using in-memory default.\"); return DEFAULT_PLAYBOOK\n", " try:\n", " with open(playbook_path, 'r') as f: playbook = json.load(f)\n", " updated = any(key not in playbook for key in DEFAULT_PLAYBOOK)\n", " if updated:\n", " logger.info(\"Updating playbook with new default strategies...\")\n", " playbook.update({k: v for k, v in DEFAULT_PLAYBOOK.items() if k not in playbook})\n", " with open(playbook_path, 'w') as f: json.dump(playbook, f, indent=4)\n", " logger.info(f\"Successfully loaded dynamic playbook from {playbook_path}\"); return playbook\n", " except (json.JSONDecodeError, IOError) as e: logger.error(f\"Failed to load or parse playbook file: {e}. Using in-memory default.\"); return DEFAULT_PLAYBOOK\n", "\n", "def load_nickname_ledger(ledger_path: str) -> Dict:\n", " logger.info(\"-> Loading Nickname Ledger...\")\n", " if os.path.exists(ledger_path):\n", " try:\n", " with open(ledger_path, 'r') as f:\n", " ledger = json.load(f)\n", " logger.info(f\" - Loaded existing nickname ledger from: {ledger_path}\")\n", " return ledger\n", " except (json.JSONDecodeError, IOError) as e:\n", " logger.error(f\" - Could not read or parse nickname ledger. Creating a new one. Error: {e}\")\n", " return {}\n", "\n", "def perform_strategic_review(history: Dict, directives_path: str) -> Tuple[Dict, List[Dict]]:\n", " logger.info(\"--- STRATEGIC REVIEW: Analyzing long-term strategy health...\")\n", " health_report = {}\n", " directives = []\n", " historical_runs = history.get(\"historical_runs\", [])\n", " if len(historical_runs) < 3:\n", " logger.info(\"--- STRATEGIC REVIEW: Insufficient history for a full review.\")\n", " return health_report, directives\n", "\n", " strategy_names = set(run.get('strategy_name') for run in historical_runs if run.get('strategy_name'))\n", " for name in strategy_names:\n", " strategy_runs = [run for run in historical_runs if run.get('strategy_name') == name]\n", " if len(strategy_runs) < 3: continue\n", " failures = sum(1 for run in strategy_runs if run.get(\"final_metrics\", {}).get(\"mar_ratio\", 0) < 0.1)\n", " chronic_failure_rate = failures / len(strategy_runs)\n", " total_cycles = sum(len(run.get(\"cycle_details\", [])) for run in strategy_runs)\n", " breaker_trips = sum(sum(1 for cycle in run.get(\"cycle_details\", []) if cycle.get(\"Status\") == \"Circuit Breaker\") for run in strategy_runs)\n", " circuit_breaker_freq = (breaker_trips / total_cycles) if total_cycles > 0 else 0\n", " health_report[name] = {\"ChronicFailureRate\": f\"{chronic_failure_rate:.0%}\", \"CircuitBreakerFrequency\": f\"{circuit_breaker_freq:.0%}\", \"RunsAnalyzed\": len(strategy_runs)}\n", "\n", " recent_runs = historical_runs[-3:]\n", " if len(recent_runs) >= 3 and len(set(r.get('strategy_name') for r in recent_runs)) == 1:\n", " stagnant_strat_name = recent_runs[0].get('strategy_name')\n", " calmar_values = [r.get(\"final_metrics\", {}).get(\"mar_ratio\", 0) for r in recent_runs]\n", " if calmar_values[2] <= calmar_values[1] <= calmar_values[0]:\n", " if stagnant_strat_name in health_report: health_report[stagnant_strat_name][\"StagnationWarning\"] = True\n", " stagnation_directive = {\"action\": \"FORCE_EXPLORATION\", \"strategy\": stagnant_strat_name, \"reason\": f\"Stagnation: No improvement over last 3 runs (MAR Ratios: {[round(c, 2) for c in calmar_values]}).\"}\n", " directives.append(stagnation_directive)\n", " logger.warning(f\"--- STRATEGIC REVIEW: Stagnation detected for '{stagnant_strat_name}'. Creating directive.\")\n", "\n", " try:\n", " with open(directives_path, 'w') as f: json.dump(directives, f, indent=4)\n", " logger.info(f\"--- STRATEGIC REVIEW: Directives saved to {directives_path}\" if directives else \"--- STRATEGIC REVIEW: No new directives generated. Cleared old directives file.\")\n", " except IOError as e: logger.error(f\"--- STRATEGIC REVIEW: Failed to write to directives file: {e}\")\n", "\n", " if health_report: logger.info(f\"--- STRATEGIC REVIEW: Health report generated.\\n{json.dumps(health_report, indent=2)}\")\n", " return health_report, directives\n", "\n", "def determine_timeframe_roles(detected_tfs: List[str]) -> Dict[str, Optional[str]]:\n", " if not detected_tfs: raise ValueError(\"No timeframes were detected from data files.\")\n", " tf_with_values = sorted([(tf, FeatureEngineer.TIMEFRAME_MAP.get(tf.upper(), 99999)) for tf in detected_tfs], key=lambda x: x[1])\n", " sorted_tfs = [tf[0] for tf in tf_with_values]\n", " roles = {'base': sorted_tfs[0], 'medium': None, 'high': None}\n", " if len(sorted_tfs) == 2: roles['high'] = sorted_tfs[1]\n", " elif len(sorted_tfs) >= 3:\n", " roles['medium'] = sorted_tfs[1]\n", " roles['high'] = sorted_tfs[2]\n", " if len(sorted_tfs) > 3: logger.warning(f\"Detected {len(sorted_tfs)} timeframes. Using {roles['base']}, {roles['medium']}, and {roles['high']}.\")\n", " logger.info(f\"Dynamically determined timeframe roles: {roles}\")\n", " return roles\n", "\n", "def run_single_instance(fallback_config: Dict, framework_history: Dict, playbook: Dict, nickname_ledger: Dict, directives: List[Dict], api_interval_seconds: int, data_source_path: str):\n", " MODEL_QUALITY_THRESHOLD = 0.05\n", " run_timestamp_str = datetime.now().strftime(\"%Y%m%d-%H%M%S\")\n", " gemini_analyzer = GeminiAnalyzer()\n", " api_timer = APITimer(interval_seconds=api_interval_seconds)\n", "\n", " current_config_dict = fallback_config.copy()\n", " current_config_dict['run_timestamp'] = run_timestamp_str\n", "\n", " temp_config_for_loader = ConfigModel(**{**current_config_dict, 'nickname': '', 'run_timestamp': 'temp', 'BASE_PATH': data_source_path})\n", " data_loader = DataLoader(temp_config_for_loader, data_source_path)\n", "\n", " all_files = [f for f in os.listdir(data_source_path) if f.endswith(('.csv', '.txt')) and re.match(r'^[A-Z0-9]+_[A-Z0-9]+', f)]\n", " if not all_files:\n", " logger.critical(f\"No data files found in the data source path ('{data_source_path}') matching pattern. Exiting.\")\n", " return\n", "\n", " data_by_tf, detected_timeframes = data_loader.load_and_parse_data(all_files)\n", " if not data_by_tf: return\n", "\n", " # Use the local workspace path for feature engineering config\n", " temp_config_for_fe = ConfigModel(**{**current_config_dict, 'nickname': '', 'run_timestamp': 'temp'})\n", " tf_roles = determine_timeframe_roles(detected_timeframes)\n", " fe = FeatureEngineer(temp_config_for_fe, tf_roles)\n", " full_df = fe.create_feature_stack(data_by_tf)\n", " if full_df.empty:\n", " logger.critical(\"Feature engineering resulted in an empty dataframe. Exiting.\")\n", " return\n", "\n", " data_summary = {}\n", " summary_df = full_df.reset_index()\n", " assets = summary_df['Symbol'].unique().tolist()\n", " data_summary['assets_detected'] = assets\n", " data_summary['time_range'] = {'start': summary_df['Timestamp'].min().isoformat(), 'end': summary_df['Timestamp'].max().isoformat()}\n", " data_summary['timeframes_used'] = tf_roles\n", " asset_stats = {asset: {'avg_atr': round(full_df[full_df['Symbol'] == asset]['ATR'].mean(), 5), 'avg_adx': round(full_df[full_df['Symbol'] == asset]['ADX'].mean(), 2), 'trending_pct': f\"{round(full_df[full_df['Symbol'] == asset]['market_regime'].mean() * 100, 1)}%\"} for asset in assets}\n", " data_summary['asset_statistics'] = asset_stats\n", " if len(assets) > 1:\n", " pivot_df = full_df.pivot(columns='Symbol', values='Close').ffill().bfill()\n", " data_summary['asset_correlation_matrix'] = pivot_df.corr().round(3).to_dict()\n", "\n", " script_name = \"End_To_End_Advanced_ML_Trading_Framework_PRO_V184_Colab.py\"\n", " version_label = script_name.replace(\".py\", \"\")\n", "\n", " # Directives path points to local workspace\n", " health_report, _ = perform_strategic_review(framework_history, fallback_config['DIRECTIVES_FILE_PATH'])\n", "\n", " ai_setup = api_timer.call(gemini_analyzer.get_initial_run_setup, version_label, nickname_ledger, framework_history, playbook, health_report, directives, data_summary)\n", "\n", " if not ai_setup:\n", " logger.critical(\"AI-driven setup failed. Using fallback configuration and exiting.\")\n", " return\n", "\n", " ai_setup = _sanitize_ai_suggestions(ai_setup)\n", "\n", " if 'RETRAINING_FREQUENCY' in ai_setup:\n", " ai_setup['RETRAINING_FREQUENCY'] = _sanitize_frequency_string(ai_setup['RETRAINING_FREQUENCY'])\n", "\n", " current_config_dict.update(ai_setup)\n", "\n", " if isinstance(ai_setup.get(\"nickname\"), str) and ai_setup.get(\"nickname\"):\n", " new_nickname = ai_setup[\"nickname\"]\n", " nickname_ledger[version_label] = new_nickname\n", " # Use the local path for the nickname ledger\n", " bootstrap_config_for_path = ConfigModel(**fallback_config, run_timestamp=\"init\", nickname=\"init\")\n", " try:\n", " with open(bootstrap_config_for_path.NICKNAME_LEDGER_PATH, 'w') as f:\n", " json.dump(nickname_ledger, f, indent=4)\n", " logger.info(f\" - Saved new nickname '{new_nickname}' to local ledger.\")\n", " except IOError as e:\n", " logger.error(f\" - Failed to save the new nickname to the ledger: {e}\")\n", "\n", " current_config_dict['REPORT_LABEL'] = version_label\n", " current_config_dict['nickname'] = nickname_ledger.get(version_label, f\"Run-{run_timestamp_str}\")\n", " config = ConfigModel(**current_config_dict)\n", "\n", " if not config.selected_features and config.strategy_name in playbook:\n", " config.selected_features = playbook[config.strategy_name].get(\"features\", [])\n", " if config.selected_features:\n", " logger.warning(f\"AI response for initial setup was missing 'selected_features'. Using default from playbook for '{config.strategy_name}'.\")\n", " else:\n", " logger.critical(f\"FATAL: AI did not provide features and no defaults exist for '{config.strategy_name}'.\")\n", " return\n", "\n", " context_features_check = [f for f in config.selected_features if '_ctx_' in f]\n", " if not context_features_check and not playbook.get(config.strategy_name, {}).get('requires_gnn'):\n", " logger.warning(f\"AI-selected features for '{config.strategy_name}' do not contain any multi-timeframe context features.\")\n", "\n", "\n", " file_handler = RotatingFileHandler(config.LOG_FILE_PATH, maxBytes=5*1024*1024, backupCount=2)\n", " file_handler.setLevel(logging.DEBUG)\n", " file_formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')\n", " file_handler.setFormatter(file_formatter)\n", " logger.addHandler(file_handler)\n", "\n", " logger.info(f\"--- Run Initialized: {config.nickname} | Strategy: {config.strategy_name} ---\")\n", " if config.analysis_notes: logger.info(f\"--- AI Analysis Notes: {config.analysis_notes} ---\")\n", "\n", " all_available_features = [c for c in full_df.columns if c not in ['Open','High','Low','Close','RealVolume','Symbol','Timestamp','primary_model_signal','target']]\n", " start_date, end_date = full_df.index.min(), full_df.index.max()\n", " train_window, forward_gap = pd.to_timedelta(config.TRAINING_WINDOW), pd.to_timedelta(config.FORWARD_TEST_GAP)\n", "\n", " retrain_freq_str = _sanitize_frequency_string(config.RETRAINING_FREQUENCY)\n", " test_start_date = start_date + train_window + forward_gap\n", " retraining_dates = pd.date_range(start=test_start_date, end=end_date, freq=retrain_freq_str)\n", "\n", " total_cycles = len(retraining_dates)\n", " logger.info(f\"Walk-forward analysis will run for {total_cycles} cycles with a retraining frequency of {retrain_freq_str}.\")\n", "\n", " aggregated_trades, aggregated_equity_curve = pd.DataFrame(), pd.Series([config.INITIAL_CAPITAL])\n", " in_run_historical_cycles, aggregated_shap = [], pd.DataFrame()\n", " last_equity, quarantine_list = config.INITIAL_CAPITAL, []\n", " probationary_strategy: Optional[str] = None\n", " consecutive_failures_on_probation = 0\n", "\n", " run_summary = {\"script_version\": config.REPORT_LABEL, \"nickname\": config.nickname, \"strategy_name\": config.strategy_name, \"run_start_ts\": config.run_timestamp, \"initial_params\": config.model_dump(mode='json')}\n", " aggregated_daily_dd_reports = []\n", "\n", " original_retraining_dates = list(retraining_dates)\n", " cycle_num = 0\n", " short_cycle_streak = 0\n", " is_forcing_long_cycle = False\n", " cycle_retry_count = 0\n", "\n", " while cycle_num < len(original_retraining_dates):\n", " period_start_date = original_retraining_dates[cycle_num]\n", " logger.info(f\"\\n--- Starting Cycle [{cycle_num + 1}/{len(original_retraining_dates)}] ---\")\n", " cycle_start_time = time.time()\n", "\n", " regime_window_end = period_start_date - forward_gap\n", " regime_window_start = regime_window_end - pd.Timedelta(days=30)\n", " df_regime = full_df.loc[regime_window_start:regime_window_end]\n", " if not df_regime.empty:\n", " regime_summary = {\n", " \"average_adx\": round(df_regime['ADX'].mean(), 2),\n", " \"average_volatility_rank\": round(df_regime['market_volatility_index'].mean(), 2),\n", " \"trending_percentage\": f\"{round(df_regime['market_regime'].mean() * 100, 1)}%\",\n", " \"asset_correlations\": df_regime.pivot(columns='Symbol', values='Close').ffill().corr().round(3).to_dict() if len(df_regime['Symbol'].unique()) > 1 else \"Single asset\"\n", " }\n", " regime_switch_suggestion = api_timer.call(gemini_analyzer.propose_regime_based_strategy_switch, regime_summary, playbook, config.strategy_name, quarantine_list)\n", " if regime_switch_suggestion and \"strategy_name\" in regime_switch_suggestion:\n", " new_strat = regime_switch_suggestion['strategy_name']\n", " logger.info(f\"!! REGIME SHIFT DETECTED !! AI recommends switching from '{config.strategy_name}' to '{new_strat}'.\")\n", " config = ConfigModel(**{**config.model_dump(mode='json'), **playbook[new_strat], **regime_switch_suggestion})\n", "\n", " retrain_freq_str = _sanitize_frequency_string(config.RETRAINING_FREQUENCY)\n", " is_short_cycle = 'M' not in retrain_freq_str.upper() and 'Y' not in retrain_freq_str.upper() and pd.to_timedelta(retrain_freq_str) < pd.Timedelta(days=60)\n", "\n", " if is_short_cycle and not is_forcing_long_cycle: short_cycle_streak += 1\n", " else: short_cycle_streak = 0\n", "\n", " if short_cycle_streak >= 4:\n", " logger.warning(\"Forced Exploration Triggered: AI has suggested short cycles for 4 consecutive periods. Overriding to test a longer cycle.\")\n", " config.RETRAINING_FREQUENCY = '90D'; retrain_freq_str = '90D'\n", " is_forcing_long_cycle = True; short_cycle_streak = 0\n", "\n", " retrain_offset = pd.tseries.frequencies.to_offset(retrain_freq_str)\n", " train_end = period_start_date - forward_gap\n", " train_start = train_end - train_window\n", " test_end = period_start_date + retrain_offset\n", " if test_end > end_date: test_end = end_date\n", "\n", " df_train_raw, df_test = full_df.loc[train_start:train_end].copy(), full_df.loc[period_start_date:test_end].copy()\n", " if df_train_raw.empty or df_test.empty:\n", " logger.warning(f\" - No data for cycle [{cycle_num + 1}/{len(original_retraining_dates)}]. Skipping.\"); cycle_num += 1; continue\n", "\n", " logger.info(f\" - Dates | Train: {train_start.date()}-{train_end.date()} | Test: {period_start_date.date()}-{test_end.date()}\")\n", "\n", " strategy_details = playbook.get(config.strategy_name, {})\n", " is_meta_model = strategy_details.get(\"requires_meta_labeling\", False)\n", "\n", " fe.config = config\n", " df_train_labeled = fe.label_meta_outcomes(df_train_raw, config.LOOKAHEAD_CANDLES) if is_meta_model else fe.label_outcomes(df_train_raw, config.LOOKAHEAD_CANDLES)\n", " if df_train_labeled.empty or ('target' in df_train_labeled and df_train_labeled['target'].abs().sum() == 0):\n", " logger.warning(\" - No valid labels generated for this training cycle. Skipping.\"); cycle_num += 1; continue\n", "\n", " config.selected_features = [f for f in config.selected_features if f in all_available_features]\n", " trainer = ModelTrainer(config)\n", " feature_list_to_use = config.selected_features if not strategy_details.get(\"requires_gnn\") else trainer.GNN_BASE_FEATURES\n", " train_result = trainer.train(df_train_labeled, feature_list_to_use, strategy_details)\n", "\n", " best_objective_score = trainer.study.best_value if trainer.study and trainer.study.best_value is not None else -1\n", "\n", " if not train_result or best_objective_score < MODEL_QUALITY_THRESHOLD:\n", " cycle_retry_count += 1\n", " logger.critical(f\"!! MODEL QUALITY GATE FAILED !! Objective score ({best_objective_score:.3f}) is below threshold ({MODEL_QUALITY_THRESHOLD}). Retry {cycle_retry_count}/{config.MAX_TRAINING_RETRIES_PER_CYCLE}.\")\n", "\n", " if cycle_retry_count > config.MAX_TRAINING_RETRIES_PER_CYCLE:\n", " logger.error(f\"!! CYCLE ABANDONED !! Exceeded max training retries ({config.MAX_TRAINING_RETRIES_PER_CYCLE}). Skipping this period with $0 PNL.\")\n", " cycle_result = {\"StartDate\": period_start_date.date().isoformat(), \"EndDate\": test_end.date().isoformat(), \"NumTrades\": 0, \"PNL\": 0.00, \"Status\": \"TRAINING_FAILURE_LIMIT\"}\n", " in_run_historical_cycles.append(cycle_result)\n", " cycle_retry_count = 0\n", " cycle_num += 1\n", " continue\n", "\n", " logger.info(\" - Re-engaging AI for new parameters to retry this cycle...\")\n", " ai_suggestions = api_timer.call(gemini_analyzer.analyze_cycle_and_suggest_changes, in_run_historical_cycles, all_available_features, strategy_details, cycle_status=\"TRAINING_FAILURE\")\n", " if ai_suggestions:\n", " ai_suggestions = _sanitize_ai_suggestions(ai_suggestions)\n", " logger.info(f\" - AI suggests updating params for retry: {ai_suggestions}\")\n", " config = ConfigModel(**{**config.model_dump(mode='json'), **ai_suggestions})\n", " else:\n", " logger.error(\" - AI failed to provide suggestions for training failure. Retrying cycle with same parameters.\")\n", " continue\n", "\n", " cycle_retry_count = 0\n", "\n", " pipeline, threshold = train_result\n", " if trainer.shap_summary is not None:\n", " aggregated_shap = trainer.shap_summary if aggregated_shap.empty else aggregated_shap.add(trainer.shap_summary, fill_value=0)\n", "\n", " X_test = trainer._get_gnn_embeddings_for_test(df_test) if trainer.is_gnn_model else df_test[feature_list_to_use].copy().fillna(0)\n", " if not X_test.empty:\n", " probs = pipeline.predict_proba(X_test)\n", " if is_meta_model and probs.shape[1] == 2: df_test[['prob_0', 'prob_1']] = probs\n", " elif not is_meta_model and probs.shape[1] == 3: df_test[['prob_short', 'prob_hold', 'prob_long']] = probs\n", "\n", " backtester = Backtester(config)\n", " trades, equity_curve, breaker_tripped, breaker_context, daily_dd_report = backtester.run_backtest_chunk(df_test, threshold, last_equity, is_meta_model)\n", " aggregated_daily_dd_reports.append(daily_dd_report)\n", "\n", " cycle_pnl = equity_curve.iloc[-1] - last_equity if not equity_curve.empty else 0\n", " cycle_result = {\"StartDate\": period_start_date.date().isoformat(), \"EndDate\": test_end.date().isoformat(), \"NumTrades\": len(trades), \"PNL\": round(cycle_pnl, 2), \"Status\": \"Circuit Breaker\" if breaker_tripped else \"Completed\"}\n", " if breaker_tripped: cycle_result[\"BreakerContext\"] = breaker_context\n", " in_run_historical_cycles.append(cycle_result)\n", "\n", " if not trades.empty:\n", " aggregated_trades = pd.concat([aggregated_trades, trades], ignore_index=True)\n", " aggregated_equity_curve = pd.concat([aggregated_equity_curve, equity_curve.iloc[1:]], ignore_index=True)\n", " last_equity = equity_curve.iloc[-1]\n", "\n", " cycle_status_for_ai = \"COMPLETED\"\n", " if breaker_tripped:\n", " if probationary_strategy == config.strategy_name:\n", " consecutive_failures_on_probation += 1\n", " logger.warning(f\"Strategy '{config.strategy_name}' failed again while on probation ({consecutive_failures_on_probation} consecutive failures).\")\n", " if consecutive_failures_on_probation >= 2:\n", " logger.critical(f\"!! QUARANTINING STRATEGY: '{config.strategy_name}' due to repeated failures.\")\n", " if config.strategy_name not in quarantine_list: quarantine_list.append(config.strategy_name)\n", "\n", " intervention_suggestion = api_timer.call(gemini_analyzer.propose_strategic_intervention, in_run_historical_cycles[-2:], playbook, config.strategy_name, quarantine_list)\n", " if intervention_suggestion and intervention_suggestion.get(\"strategy_name\") in playbook:\n", " intervention_suggestion = _sanitize_ai_suggestions(intervention_suggestion)\n", " logger.info(f\" - Strategic Intervention successful. Switching to strategy: {intervention_suggestion['strategy_name']}\")\n", " config = ConfigModel(**{**config.model_dump(mode='json'), **intervention_suggestion})\n", " probationary_strategy, consecutive_failures_on_probation = None, 0\n", " else:\n", " logger.error(\" - Strategic intervention FAILED. Halting run to prevent further losses.\")\n", " break\n", " else:\n", " cycle_status_for_ai = \"PROBATION\"\n", " else:\n", " logger.warning(f\"Strategy '{config.strategy_name}' hit a circuit breaker. Placing it on PROBATION.\")\n", " probationary_strategy = config.strategy_name\n", " consecutive_failures_on_probation = 1\n", " cycle_status_for_ai = \"PROBATION\"\n", " else:\n", " if probationary_strategy == config.strategy_name:\n", " logger.info(f\"Strategy '{config.strategy_name}' completed a cycle successfully and is now OFF probation.\")\n", " probationary_strategy, consecutive_failures_on_probation = None, 0\n", " cycle_status_for_ai = \"COMPLETED\"\n", "\n", " suggested_params = api_timer.call(gemini_analyzer.analyze_cycle_and_suggest_changes, in_run_historical_cycles, all_available_features, strategy_details, cycle_status=cycle_status_for_ai)\n", " if suggested_params:\n", " suggested_params = _sanitize_ai_suggestions(suggested_params)\n", " logger.info(f\" - AI suggests updating params: {suggested_params}\")\n", "\n", " if 'selected_features' in suggested_params:\n", " suggested_params['selected_features'] = [f for f in suggested_params['selected_features'] if f in all_available_features]\n", "\n", " if 'RETRAINING_FREQUENCY' in suggested_params:\n", " if is_forcing_long_cycle:\n", " logger.warning(\" - Ignoring AI's frequency suggestion this cycle to enforce the longer test period.\")\n", " del suggested_params['RETRAINING_FREQUENCY']\n", " else:\n", " suggested_params['RETRAINING_FREQUENCY'] = _sanitize_frequency_string(suggested_params['RETRAINING_FREQUENCY'])\n", "\n", " config = ConfigModel(**{**config.model_dump(mode='json'), **suggested_params})\n", "\n", " if is_forcing_long_cycle: is_forcing_long_cycle = False\n", "\n", " cycle_num += 1\n", " logger.info(f\"--- Cycle complete. PNL: ${cycle_pnl:,.2f} | Final Equity: ${last_equity:,.2f} | Time: {time.time() - cycle_start_time:.2f}s ---\")\n", "\n", " final_metrics = {}\n", " if not aggregated_trades.empty:\n", " pa = PerformanceAnalyzer(config)\n", " final_metrics = pa.generate_full_report(aggregated_trades, aggregated_equity_curve, in_run_historical_cycles, aggregated_shap, framework_history, aggregated_daily_dd_reports)\n", "\n", " run_summary.update({\"run_end_ts\": datetime.now().strftime(\"%Y%m%d-%H%M%S\"), \"final_metrics\": final_metrics, \"cycle_details\": in_run_historical_cycles, \"top_5_features\": aggregated_shap.head(5).index.tolist() if not aggregated_shap.empty else [], \"final_params\": config.model_dump(mode='json')})\n", " save_run_to_memory(config, run_summary, framework_history)\n", "\n", " logger.removeHandler(file_handler)\n", " file_handler.close()\n", "\n", "def setup_local_workspace(drive_path: str, local_path: str) -> None:\n", " \"\"\"Creates a local workspace and copies essential files from Drive.\"\"\"\n", " logger.info(f\"--- Setting up local workspace at '{local_path}' ---\")\n", " if os.path.exists(local_path):\n", " shutil.rmtree(local_path)\n", " os.makedirs(os.path.join(local_path, \"Results\"), exist_ok=True)\n", "\n", " files_to_copy = [\n", " \"Results/champion.json\",\n", " \"Results/historical_runs.jsonl\",\n", " \"Results/strategy_playbook.json\",\n", " \"Results/nickname_ledger.json\",\n", " \"Results/framework_directives.json\",\n", " \".env\"\n", " ]\n", " for file_rel_path in files_to_copy:\n", " source_file = os.path.join(drive_path, file_rel_path)\n", " dest_file = os.path.join(local_path, file_rel_path)\n", " if os.path.exists(source_file):\n", " os.makedirs(os.path.dirname(dest_file), exist_ok=True)\n", " shutil.copy2(source_file, dest_file)\n", " logger.info(f\" - Copied '{file_rel_path}' to local workspace.\")\n", " else:\n", " logger.warning(f\" - Optional file not found in Drive, will be created if needed: '{file_rel_path}'\")\n", "\n", "def sync_results_to_drive(local_path: str, drive_path: str) -> None:\n", " \"\"\"Copies all generated results from the local workspace back to Drive.\"\"\"\n", " logger.info(f\"--- Syncing results from '{local_path}' to '{drive_path}' ---\")\n", " try:\n", " # This will copy the entire local 'Results' directory to the Drive path\n", " source_results = os.path.join(local_path, \"Results\")\n", " dest_results = os.path.join(drive_path, \"Results\")\n", " if os.path.exists(source_results):\n", " # The dirs_exist_ok=True argument for copytree is available in Python 3.8+\n", " shutil.copytree(source_results, dest_results, dirs_exist_ok=True)\n", " logger.info(\"[SUCCESS] All results have been synced back to Google Drive.\")\n", " except Exception as e:\n", " logger.error(f\"FATAL: Failed to sync results back to Google Drive: {e}\", exc_info=True)\n", "\n", "\n", "def main():\n", " # --- IMPORTANT COLAB CONFIGURATION ---\n", " DRIVE_PERSISTENT_PATH = \"/content/drive/MyDrive/TradingData\"\n", " LOCAL_WORKSPACE_PATH = \"/content/trading_workspace\"\n", " # -------------------------------------\n", "\n", " # Setup local workspace and handle teardown\n", " setup_local_workspace(DRIVE_PERSISTENT_PATH, LOCAL_WORKSPACE_PATH)\n", "\n", " try:\n", " # Load .env from the LOCAL workspace now\n", " load_dotenv(dotenv_path=os.path.join(LOCAL_WORKSPACE_PATH, '.env'))\n", "\n", " CONTINUOUS_RUN_HOURS = 0; MAX_RUNS = 1\n", " fallback_config={\n", " # BASE_PATH now points to the LOCAL workspace for all operations\n", " \"BASE_PATH\": LOCAL_WORKSPACE_PATH,\n", " \"REPORT_LABEL\": \"ML_Framework_V184_Regime_Intel\",\n", " \"strategy_name\": \"Meta_Labeling_Filter\", \"INITIAL_CAPITAL\": 100000.0,\n", " \"CONFIDENCE_TIERS\": {'ultra_high':{'min':0.8,'risk_mult':1.2,'rr':3.0},'high':{'min':0.7,'risk_mult':1.0,'rr':2.5},'standard':{'min':0.6,'risk_mult':0.8,'rr':2.0}},\n", " \"BASE_RISK_PER_TRADE_PCT\": 0.01,\"RISK_CAP_PER_TRADE_USD\": 5000.0,\n", " \"SPREAD_PCTG_OF_ATR\": 0.05, \"SLIPPAGE_PCTG_OF_ATR\": 0.02,\n", " \"OPTUNA_TRIALS\": 50, \"TRAINING_WINDOW\": '365D', \"RETRAINING_FREQUENCY\": '90D',\n", " \"FORWARD_TEST_GAP\": \"1D\", \"LOOKAHEAD_CANDLES\": 150, \"TREND_FILTER_THRESHOLD\": 25.0,\n", " \"BOLLINGER_PERIOD\": 20, \"STOCHASTIC_PERIOD\": 14, \"CALCULATE_SHAP_VALUES\": True,\n", " \"MAX_DD_PER_CYCLE\": 0.25,\"GNN_EMBEDDING_DIM\": 8, \"GNN_EPOCHS\": 50,\n", " \"MIN_VOLATILITY_RANK\": 0.1, \"MAX_VOLATILITY_RANK\": 0.9,\n", " \"selected_features\": [],\n", " \"MAX_CONCURRENT_TRADES\": 3,\n", " \"USE_PARTIAL_PROFIT\": False,\n", " \"PARTIAL_PROFIT_TRIGGER_R\": 1.5,\n", " \"PARTIAL_PROFIT_TAKE_PCT\": 0.5,\n", " \"MAX_TRAINING_RETRIES_PER_CYCLE\": 3\n", " }\n", "\n", " # The directives file path also points to the local workspace\n", " fallback_config[\"DIRECTIVES_FILE_PATH\"] = os.path.join(fallback_config[\"BASE_PATH\"], \"Results\", \"framework_directives.json\")\n", "\n", " api_interval_seconds = 300\n", "\n", " run_count = 0; script_start_time = datetime.now(); is_continuous = CONTINUOUS_RUN_HOURS > 0 or MAX_RUNS > 1\n", "\n", " # Bootstrap config also uses the local path\n", " bootstrap_config = ConfigModel(**fallback_config, run_timestamp=\"init\", nickname=\"init\")\n", " playbook = initialize_playbook(bootstrap_config.BASE_PATH)\n", "\n", " while True:\n", " run_count += 1\n", " if is_continuous: logger.info(f\"\\n{'='*30} STARTING DAEMON RUN {run_count} {'='*30}\\n\")\n", " else: logger.info(f\"\\n{'='*30} STARTING SINGLE RUN {'='*30}\\n\")\n", "\n", " # Load memory files from the local workspace\n", " nickname_ledger = load_nickname_ledger(bootstrap_config.NICKNAME_LEDGER_PATH)\n", " framework_history = load_memory(bootstrap_config.CHAMPION_FILE_PATH, bootstrap_config.HISTORY_FILE_PATH)\n", "\n", " directives = []\n", " if os.path.exists(bootstrap_config.DIRECTIVES_FILE_PATH):\n", " try:\n", " with open(bootstrap_config.DIRECTIVES_FILE_PATH, 'r') as f: directives = json.load(f)\n", " if directives: logger.info(f\"Loaded {len(directives)} directive(s) for this run.\")\n", " except (json.JSONDecodeError, IOError) as e: logger.error(f\"Could not load directives file: {e}\")\n", "\n", " try:\n", " # The data source path is the persistent Drive path\n", " run_single_instance(fallback_config, framework_history, playbook, nickname_ledger, directives, api_interval_seconds, data_source_path=DRIVE_PERSISTENT_PATH)\n", " except Exception as e:\n", " logger.critical(f\"A critical, unhandled error occurred during run {run_count}: {e}\", exc_info=True)\n", " if not is_continuous: break\n", " logger.info(\"Attempting to continue after a 60-second cooldown...\"); time.sleep(60)\n", "\n", " if not is_continuous:\n", " logger.info(\"Single run complete.\")\n", " break\n", "\n", " if MAX_RUNS > 1 and run_count >= MAX_RUNS:\n", " logger.info(f\"Reached max run limit of {MAX_RUNS}. Exiting daemon mode.\")\n", " break\n", "\n", " if CONTINUOUS_RUN_HOURS > 0 and (datetime.now() - script_start_time).total_seconds() / 3600 >= CONTINUOUS_RUN_HOURS:\n", " logger.info(f\"Reached max runtime of {CONTINUOUS_RUN_HOURS} hours. Exiting daemon mode.\")\n", " break\n", "\n", " try:\n", " sys.stdout.write(\"\\n\")\n", " for i in range(10, 0, -1):\n", " sys.stdout.write(f\"\\r>>> Run {run_count} complete. Press Ctrl+C to stop. Continuing in {i:2d} seconds...\"); sys.stdout.flush(); time.sleep(1)\n", " sys.stdout.write(\"\\n\\n\")\n", " except KeyboardInterrupt:\n", " logger.info(\"\\n\\nDaemon stopped by user. Exiting gracefully.\")\n", " break\n", " finally:\n", " # This block ensures results are synced back to Drive even if the script fails\n", " sync_results_to_drive(LOCAL_WORKSPACE_PATH, DRIVE_PERSISTENT_PATH)" ], "metadata": { "id": "jMXGvXCe93pF", "colab": { "base_uri": "https://localhost:8080/" }, "outputId": "def803ab-ad7a-4394-a495-3d3cc5df5e36" }, "execution_count": 16, "outputs": [ { "output_type": "stream", "name": "stdout", "text": [ "2025-06-12 15:37:15,051 - INFO - PyTorch and PyG loaded successfully. GNN module is available.\n" ] }, { "output_type": "stream", "name": "stderr", "text": [ "INFO:ML_Trading_Framework:PyTorch and PyG loaded successfully. GNN module is available.\n" ] } ] }, { "cell_type": "code", "source": [ "if __name__ == '__main__':\n", " main()" ], "metadata": { "id": "NlZ49gsymjb9" }, "execution_count": 18, "outputs": [] } ] }