469 lines
13 KiB
Markdown
469 lines
13 KiB
Markdown
# Python ML Pipeline Integration Guide
| |||
| |||
## Overview
| |||
| |||
The DualEA Python ML pipeline has been fully integrated. Previously orphaned modules are now properly connected and accessible through the unified `ML` package.
| |||
| |||
## Integration Status
| |||
| |||
### Previously Orphaned → Now Integrated ✅
| |||
| |||
| Module | Previous Status | Current Integration | Usage |
| |||
|--------|----------------|-------------------|-------|
| |||
| `shadow_executor.py` | Standalone script | `ML.start_shadow_executor()` | Shadow trading on demo |
| |||
| `feature_export.py` | Standalone functions | `ML.write_batches_to_parquet()` | Protobuf → Parquet |
| |||
| `feature_export_dll.py` | Unreferenced | `ML.export_feature_batch()` | DLL export |
| |||
| `delta_ingest.py` | Standalone script | `ML.ingest_to_delta()` | Delta Lake ingestion |
| |||
| `online_learner.py` | Partially used | `ML.OnlineLearner` | Incremental learning |
| |||
| `continuous_trainer.py` | Standalone script | `ML.run_full_pipeline()` | Automated retraining |
| |||
| |||
## Quick Start
| |||
| |||
### Import All Modules
| |||
| |||
```python
| |||
from ML import (
| |||
# Core training
| |||
train,
| |||
NewsFetcher,
| |||
export_policy,
| |||
| |||
# Previously orphaned - now integrated
| |||
write_batches_to_parquet,
| |||
ingest_to_delta,
| |||
OnlineLearner,
| |||
start_shadow_executor,
| |||
run_full_pipeline
| |||
)
| |||
```
| |||
| |||
### Use Integration Runner
| |||
| |||
```bash
| |||
# Run full pipeline with all services
| |||
python -m ML.integration_runner --mode all-services \
| |||
--account 123456 --password secret --server broker-Demo
| |||
| |||
# Run single service
| |||
python -m ML.integration_runner --mode shadow \
| |||
--account 123456 --password secret --server broker-Demo
| |||
| |||
# Run feature export service
| |||
python -m ML.integration_runner --mode feature-export
| |||
```
| |||
| |||
## Module Details
| |||
| |||
### 1. Shadow Executor (`shadow_executor.py`)
| |||
| |||
**Purpose:** Execute trades on demo account to shadow live trading
| |||
| |||
**Integration:**
| |||
```python
| |||
from ML import start_shadow_executor
| |||
| |||
# Start shadow trading
| |||
start_shadow_executor(
| |||
demo_account=123456,
| |||
demo_password='secret',
| |||
demo_server='broker-Demo'
| |||
)
| |||
```
| |||
| |||
**Features:**
| |||
- Reads orders from `DualEA/shadow/orders_pending.csv`
| |||
- Executes on MT5 demo account
| |||
- Records results to `DualEA/shadow/shadow_results.csv`
| |||
- Calculates R-multiples for performance tracking
| |||
| |||
**Files Used:**
| |||
- Input: `orders_pending.csv`
| |||
- Output: `orders_executed.csv`, `shadow_results.csv`
| |||
| |||
### 2. Feature Export (`feature_export.py`)
| |||
| |||
**Purpose:** Convert protobuf feature batches to Parquet format
| |||
| |||
**Integration:**
| |||
```python
| |||
from ML import write_batches_to_parquet
| |||
| |||
# Export protobuf bytes to Parquet
| |||
with open('features.pb', 'rb') as f:
| |||
pb_bytes = f.read()
| |||
| |||
output_path = write_batches_to_parquet(pb_bytes, fname='features.parquet')
| |||
```
| |||
| |||
**Features:**
| |||
- Converts `FeatureBatchEnvelope` protobuf to PyArrow
| |||
- Writes compressed Parquet files
| |||
- Automatic retry with CSV fallback
| |||
- Batch processing support
| |||
| |||
**Configuration:**
| |||
```python
| |||
# Default export directory
| |||
EXPORT_DIR = r"C:\DualEA_FeatureBatches"
| |||
```
| |||
| |||
### 3. Delta Lake Ingestion (`delta_ingest.py`)
| |||
| |||
**Purpose:** Ingest Parquet files to Delta Lake for analytics
| |||
| |||
**Integration:**
| |||
```python
| |||
from ML import ingest_to_delta
| |||
| |||
# Ingest all Parquet files to Delta Lake
| |||
ingest_to_delta()
| |||
```
| |||
| |||
**Features:**
| |||
- Discovers all `.parquet` files in feature directory
| |||
- Appends to Delta Lake table
| |||
- Schema evolution support
| |||
- Time travel capabilities
| |||
| |||
**Configuration:**
| |||
```python
| |||
FEATURE_DIR = r"C:\DualEA_FeatureBatches"
| |||
DELTA_DIR = r"C:\DualEA_DeltaLake"
| |||
```
| |||
| |||
### 4. Online Learner (`online_learner.py`)
| |||
| |||
**Purpose:** Incremental model updates as trades complete
| |||
| |||
**Integration:**
| |||
```python
| |||
from ML import OnlineLearner
| |||
| |||
# Initialize online learner
| |||
learner = OnlineLearner(
| |||
model_type='classifier',
| |||
learning_rate=0.01
| |||
)
| |||
| |||
# Update with new trade
| |||
learner.partial_fit(features, outcome)
| |||
| |||
# Get updated predictions
| |||
probability = learner.predict(features)
| |||
```
| |||
| |||
**Features:**
| |||
- SGDClassifier/SGDRegressor for online learning
| |||
- Keras model support (if available)
| |||
- Automatic feature scaling
| |||
- Model versioning
| |||
| |||
### 5. Integration Runner (`integration_runner.py`)
| |||
| |||
**Purpose:** Unified service manager for all ML infrastructure
| |||
| |||
**Usage:**
| |||
```bash
| |||
# Run all services
| |||
python -m ML.integration_runner --mode all-services
| |||
| |||
# Available modes:
| |||
# - full-pipeline: Single training run
| |||
# - shadow: Shadow executor only
| |||
# - feature-export: Feature export service
| |||
# - delta-ingest: Delta Lake ingestion
| |||
# - continuous-train: Continuous retraining
| |||
# - news-fetch: Economic calendar updates
| |||
# - all-services: Everything in parallel
| |||
```
| |||
| |||
**Service Architecture:**
| |||
```
| |||
┌─────────────────────────────────────────────────────────────┐
| |||
│ Service Manager │
| |||
├─────────────────────────────────────────────────────────────┤
| |||
│ │
| |||
│ ┌──────────────────┐ ┌──────────────────┐ │
| |||
│ │ Shadow Executor │ │ Feature Export │ │
| |||
│ │ (MT5 Demo) │ │ (Protobuf→Parq) │ │
| |||
│ └──────────────────┘ └──────────────────┘ │
| |||
│ │
| |||
│ ┌──────────────────┐ ┌──────────────────┐ │
| |||
│ │ Delta Ingest │ │ Continuous Train │ │
| |||
│ │ (Delta Lake) │ │ (Auto-retrain) │ │
| |||
│ └──────────────────┘ └──────────────────┘ │
| |||
│ │
| |||
│ ┌──────────────────────────────────────────┐ │
| |||
│ │ News Fetcher │ │
| |||
│ │ (Economic Calendar Updates) │ │
| |||
│ └──────────────────────────────────────────┘ │
| |||
│ │
| |||
└─────────────────────────────────────────────────────────────┘
| |||
```
| |||
| |||
## Package Structure
| |||
| |||
```
| |||
ML/
| |||
├── __init__.py # Package initialization, exports all modules
| |||
├── integration_runner.py # Master service manager
| |||
│
| |||
├── Core Training
| |||
├── train.py
| |||
├── model.py
| |||
├── features.py
| |||
├── dataset.py
| |||
├── policy.py
| |||
├── policy_export.py
| |||
│
| |||
├── Advanced Models
| |||
├── model_pytorch.py
| |||
├── model_ensemble.py
| |||
├── train_pytorch.py
| |||
├── train_neurobook.py
| |||
│
| |||
├── Previously Orphaned (Now Integrated)
| |||
├── shadow_executor.py # ✅ Integrated via ML.start_shadow_executor()
| |||
├── feature_export.py # ✅ Integrated via ML.write_batches_to_parquet()
| |||
├── feature_export_dll.py # ✅ Integrated via ML.export_feature_batch()
| |||
├── delta_ingest.py # ✅ Integrated via ML.ingest_to_delta()
| |||
├── online_learner.py # ✅ Integrated via ML.OnlineLearner
| |||
├── continuous_trainer.py # ✅ Integrated via ML.run_full_pipeline()
| |||
│
| |||
├── Infrastructure
| |||
├── news_fetcher.py # ✅ Integrated via ML.NewsFetcher
| |||
├── experiment_runner.py
| |||
├── feature_store.py
| |||
├── feedback_watcher.py
| |||
├── gate_optimizer.py
| |||
├── policy_server.py
| |||
│
| |||
└── Advanced Training
| |||
├── snapshot_train.py
| |||
├── snapshot_dataset.py
| |||
├── snapshot_policy_export.py
| |||
└── simple_optimizer.py
| |||
```
| |||
| |||
## Full Pipeline Example
| |||
| |||
```python
| |||
from ML import run_full_pipeline, start_shadow_executor, NewsFetcher
| |||
| |||
# 1. Run full training pipeline
| |||
result = run_full_pipeline(
| |||
common_dir='DualEA',
| |||
epochs=50,
| |||
batch_size=32,
| |||
min_conf=0.55,
| |||
use_news=True
| |||
)
| |||
| |||
print(f"Training complete: {result['files']['policy']}")
| |||
| |||
# 2. Start shadow executor for validation
| |||
start_shadow_executor(
| |||
demo_account=123456,
| |||
demo_password='secret',
| |||
demo_server='broker-Demo'
| |||
)
| |||
| |||
# 3. Fetch economic calendar
| |||
fetcher = NewsFetcher()
| |||
fetcher.fetch_from_mql5_calendar()
| |||
fetcher.generate_blackout_csv()
| |||
```
| |||
| |||
## Configuration
| |||
| |||
### Environment Variables
| |||
| |||
```bash
| |||
# MT5 Common Files directory
| |||
export DUALEA_COMMON_DIR="C:/Users/.../Common/Files/DualEA"
| |||
| |||
# Feature export directory
| |||
export DUALEA_FEATURE_DIR="C:/DualEA_FeatureBatches"
| |||
| |||
# Delta Lake directory
| |||
export DUALEA_DELTA_DIR="C:/DualEA_DeltaLake"
| |||
| |||
# Shadow executor credentials (optional)
| |||
export DUALEA_DEMO_ACCOUNT="123456"
| |||
export DUALEA_DEMO_PASSWORD="secret"
| |||
export DUALEA_DEMO_SERVER="broker-Demo"
| |||
```
| |||
| |||
### Default Paths
| |||
| |||
All modules use sensible defaults that can be overridden:
| |||
| |||
```python
| |||
from pathlib import Path
| |||
import os
| |||
| |||
# Common Files (auto-detected)
| |||
DEFAULT_COMMON_DIR = os.path.join(
| |||
os.environ.get("APPDATA", ""),
| |||
"MetaQuotes", "Terminal", "Common", "Files", "DualEA"
| |||
)
| |||
| |||
# Feature Batches
| |||
EXPORT_DIR = r"C:\DualEA_FeatureBatches"
| |||
| |||
# Delta Lake
| |||
DELTA_DIR = r"C:\DualEA_DeltaLake"
| |||
| |||
# Shadow Orders
| |||
PENDING_ORDERS_FILE = Path("DualEA/shadow/orders_pending.csv")
| |||
EXECUTED_ORDERS_FILE = Path("DualEA/shadow/orders_executed.csv")
| |||
RESULTS_FILE = Path("DualEA/shadow/shadow_results.csv")
| |||
```
| |||
| |||
## Service Mode Operation
| |||
| |||
### Docker-Style Service Management
| |||
| |||
```python
| |||
from ML.integration_runner import ServiceManager
| |||
| |||
# Create manager
| |||
manager = ServiceManager()
| |||
| |||
# Start services
| |||
manager.start_service('shadow', start_shadow_executor, account, password, server)
| |||
manager.start_service('features', run_feature_export_service, poll_dir, output_dir)
| |||
manager.start_service('delta', run_delta_ingest_service, interval)
| |||
| |||
# Get status
| |||
status = manager.get_status()
| |||
print(status)
| |||
| |||
# Stop all
| |||
manager.stop_all()
| |||
```
| |||
| |||
### Background Execution
| |||
| |||
```bash
| |||
# Run as background service
| |||
nohup python -m ML.integration_runner --mode all-services > integration.log 2>&1 &
| |||
| |||
# Check status
| |||
tail -f integration.log
| |||
| |||
# Stop gracefully
| |||
kill -INT <pid>
| |||
```
| |||
| |||
## Troubleshooting
| |||
| |||
### Issue: `ModuleNotFoundError: No module named 'ML'`
| |||
| |||
**Fix:** Ensure you're running from the correct directory:
| |||
```bash
| |||
cd MQL5/Experts/Advisors/DualEA
| |||
python -m ML.integration_runner --help
| |||
```
| |||
| |||
### Issue: `ImportError: No module named 'MetaTrader5'`
| |||
| |||
**Fix:** Install MT5 package or use shadow mode offline:
| |||
```bash
| |||
pip install MetaTrader5
| |||
# OR
| |||
python -m ML.integration_runner --mode feature-export # No MT5 needed
| |||
```
| |||
| |||
### Issue: Delta Lake not available
| |||
| |||
**Fix:** Install deltalake or skip:
| |||
```bash
| |||
pip install deltalake
| |||
# OR
| |||
python -m ML.integration_runner --mode all-services --no-delta
| |||
```
| |||
| |||
### Issue: Feature export not finding protobuf files
| |||
| |||
**Fix:** Check directory structure:
| |||
```python
| |||
from pathlib import Path
| |||
poll_dir = Path("DualEA/protobuf_pending")
| |||
print(f"Looking in: {poll_dir.absolute()}")
| |||
print(f"Files found: {list(poll_dir.glob('*.pb'))}")
| |||
```
| |||
| |||
## Migration from Standalone Scripts
| |||
| |||
### Before (Standalone)
| |||
| |||
```python
| |||
# shadow_executor.py - standalone
| |||
import MetaTrader5 as mt5
| |||
# ... all logic in one file
| |||
# had to be run separately
| |||
# no integration with training pipeline
| |||
```
| |||
| |||
### After (Integrated)
| |||
| |||
```python
| |||
from ML import start_shadow_executor
| |||
| |||
# Part of unified package
| |||
# Can be called from training pipeline
| |||
# Shares configuration
| |||
# Managed by ServiceManager
| |||
start_shadow_executor(account, password, server)
| |||
```
| |||
| |||
## API Reference
| |||
| |||
### Core Functions
| |||
| |||
```python
| |||
# Run complete pipeline
| |||
ML.run_full_pipeline(common_dir, epochs, batch_size, min_conf, use_news)
| |||
| |||
# Start shadow executor
| |||
ML.start_shadow_executor(account, password, server)
| |||
| |||
# Export features to Parquet
| |||
ML.write_batches_to_parquet(pb_bytes, fname)
| |||
| |||
# Ingest to Delta Lake
| |||
ML.ingest_to_delta()
| |||
| |||
# Fetch news calendar
| |||
ML.NewsFetcher().fetch_from_mql5_calendar()
| |||
| |||
# Online learning
| |||
ML.OnlineLearner().partial_fit(X, y)
| |||
```
| |||
| |||
### Classes
| |||
| |||
```python
| |||
# Ensemble models
| |||
ML.EnsembleModel(models)
| |||
| |||
# Experiment runner
| |||
ML.ExperimentRunner(base_dir)
| |||
| |||
# Feature store
| |||
ML.FeatureStore()
| |||
| |||
# Policy management
| |||
ML.Policy(version, timestamp, min_confidence)
| |||
```
| |||
| |||
## Related Documentation
| |||
| |||
- [NewsFilter_Integration_Guide.md](NewsFilter_Integration_Guide.md) - News system
| |||
- [README.md](README.md) - Main documentation
| |||
- [ML/README.md](ML/README.md) - ML pipeline overview
| |||
| |||
---
| |||
| |||
**Integration Complete:** All previously orphaned Python modules are now properly integrated into the unified `ML` package.
|