feat(step1): Universal EA Schema Layer — replaces param_manifest.yaml

NEW FILES:
  ea/__init__.py          - EA module package
  ea/schema.py            - ParameterDef + ParameterSchema (replaces YAML manifest)
  ea/set_parser.py        - Parse ANY MT5 .set file → ParameterSchema
  ea/registry.py          - EAProfile + EARegistry (ea_registry.yaml)
  ea_registry.yaml        - LEGSTECH_EA_V2 pre-registered with automation overrides

MODIFIED:
  mt5/ini_builder.py      - Accepts ParameterSchema OR legacy manifest_path
  mt5/runner.py           - Accepts optional EAProfile for EA-agnostic validation
  config.yaml             - Removed hardcoded ea: block; added ea_registry path

KEY DESIGN:
  - .set file IS the manifest: value|min|max|step auto-detects float/int/bool/enum/fixed
  - automation_overrides in EAProfile forces headless-safe values (InpShowPanel=0)
  - Phase A optimization INI: Optimization=2 (genetic), ranges from schema
  - Phase B single backtest: Optimization=0 (unchanged behavior)
  - LEGSTECH advanced mode: fully backwards compatible via legacy manifest fallback
  - runner.py still works for legacy callers (no EAProfile needed)

TESTED:
  - 49/49 LEGSTECH params parsed correctly
  - InpShowPanel=0 automation override applied
  - INI output correct for both Phase A (Optimization=2) and Phase B (Optimization=0)
  - All imports pass
This commit is contained in:
LEGSTECH Optimizer
2026-04-13 19:55:16 +00:00
parent b70a6760ae
commit d5a8973c05
10 changed files with 878 additions and 74 deletions
+3
View File
@@ -0,0 +1,3 @@
"""
ea/ — EA Profile management, .set file parsing, and parameter schema.
"""
+207
View File
@@ -0,0 +1,207 @@
"""
ea/registry.py
EA Profile storage and retrieval.
Profiles are stored in ea_registry.yaml (path from config.yaml paths.ea_registry).
The registry is the single source of truth for which EAs are registered
and how to find their .set files.
"""
from __future__ import annotations
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
import yaml
from loguru import logger
from ea.schema import ParameterSchema
from ea.set_parser import SetParser
# ── EAProfile ─────────────────────────────────────────────────────────────────
@dataclass
class EAProfile:
"""Configuration for one EA registered in the optimizer."""
name: str # Display name, e.g. "LEGSTECH_EA_V2"
ex5_file: str # MT5 Experts file name (without .ex5 extension)
set_template: str # Absolute path string to template .set file
symbol: str # e.g. "XAUUSD"
timeframe: str # e.g. "H1"
mode: str = "generic" # "generic" | "advanced"
registered_at: str = field(
default_factory=lambda: datetime.now(timezone.utc).isoformat()
)
# Params the user has chosen to optimize (param names → True/False).
# Empty dict means: use default_optimize logic in SetParser.
optimize_params: dict[str, bool] = field(default_factory=dict)
# Automation overrides: param values that must be used during backtesting,
# regardless of what the .set template says.
# Example: {"InpShowPanel": 0, "InpTesterMode": 1}
automation_overrides: dict[str, object] = field(default_factory=dict)
def __post_init__(self):
assert self.mode in ("generic", "advanced"), \
f"EAProfile.mode must be 'generic' or 'advanced', got {self.mode!r}"
assert self.timeframe in (
"M1","M5","M15","M30","H1","H4","D1","W1","MN"
), f"Invalid timeframe: {self.timeframe}"
@property
def set_template_path(self) -> Path:
return Path(self.set_template)
def to_dict(self) -> dict:
return asdict(self)
@classmethod
def from_dict(cls, d: dict) -> "EAProfile":
return cls(**{k: v for k, v in d.items() if k in cls.__dataclass_fields__})
# ── EARegistry ────────────────────────────────────────────────────────────────
class EARegistry:
"""
Manages registered EA profiles, persisted in ea_registry.yaml.
Usage:
reg = EARegistry("config.yaml")
profile = reg.get("LEGSTECH_EA_V2")
schema = reg.get_schema(profile)
"""
def __init__(self, config_path: str | Path = "config.yaml"):
config_path = Path(config_path)
with open(config_path) as f:
cfg = yaml.safe_load(f)
registry_rel = cfg.get("paths", {}).get("ea_registry", "ea_registry.yaml")
self._registry_path = config_path.parent / registry_rel
self._parser = SetParser()
self._profiles: dict[str, EAProfile] = {}
self._load()
# ── Public API ────────────────────────────────────────────────────────────
def register(self, profile: EAProfile) -> None:
"""Add or update an EA profile."""
self._profiles[profile.name] = profile
self._save()
logger.info(f"EARegistry: registered {profile.name!r} (mode={profile.mode})")
def get(self, name: str) -> EAProfile:
"""Get a registered EA profile by name. Raises KeyError if not found."""
if name not in self._profiles:
available = list(self._profiles.keys())
raise KeyError(
f"EA {name!r} not registered. Available: {available}"
)
return self._profiles[name]
def list_all(self) -> list[EAProfile]:
"""Return all registered profiles."""
return list(self._profiles.values())
def remove(self, name: str) -> None:
"""Unregister an EA."""
self._profiles.pop(name, None)
self._save()
logger.info(f"EARegistry: removed {name!r}")
def exists(self, name: str) -> bool:
return name in self._profiles
def get_schema(
self,
profile: EAProfile,
apply_optimize_selection: bool = True,
) -> ParameterSchema:
"""
Parse the EA's .set file and return a ParameterSchema.
Applies automation_overrides to ensure headless-safe defaults.
Applies optimize_params selection if present.
"""
if not profile.set_template_path.exists():
raise FileNotFoundError(
f"Set template not found: {profile.set_template}\n"
f"Please update the path in EA Registry for {profile.name!r}."
)
schema = self._parser.parse(
path=profile.set_template_path,
ea_name=profile.name,
default_optimize=False,
)
# Apply automation overrides: force specific param values
# (e.g. InpShowPanel=0 so no GUI renders during headless backtests)
for pname, value in profile.automation_overrides.items():
if pname in schema.parameters:
schema.parameters[pname].default = value
schema.parameters[pname].type = "fixed"
schema.parameters[pname].optimize = False
if apply_optimize_selection and profile.optimize_params:
for pname, should_opt in profile.optimize_params.items():
if pname in schema.parameters:
p = schema.parameters[pname]
if p.type != "fixed":
p.optimize = should_opt
elif not profile.optimize_params:
# No selection yet — mark all non-fixed as optimizable by default
for p in schema.parameters.values():
if p.type != "fixed":
p.optimize = True
return schema
def update_optimize_params(self, name: str, optimize_params: dict[str, bool]) -> None:
"""Update which parameters to optimize for a registered EA."""
profile = self.get(name)
profile.optimize_params = optimize_params
self._save()
# ── Internal ─────────────────────────────────────────────────────────────
def _load(self) -> None:
"""Load profiles from YAML. Creates the file if it doesn't exist."""
if not self._registry_path.exists():
logger.info(f"EARegistry: creating new registry at {self._registry_path}")
self._registry_path.write_text("profiles: []\n", encoding="utf-8")
return
with open(self._registry_path, encoding="utf-8") as f:
data = yaml.safe_load(f) or {}
profiles_raw = data.get("profiles", [])
for item in profiles_raw:
try:
p = EAProfile.from_dict(item)
self._profiles[p.name] = p
except Exception as e:
logger.warning(f"EARegistry: skipped malformed profile {item}: {e}")
logger.info(f"EARegistry: loaded {len(self._profiles)} profile(s) from {self._registry_path.name}")
def _save(self) -> None:
"""Persist all profiles to YAML."""
data = {"profiles": [p.to_dict() for p in self._profiles.values()]}
with open(self._registry_path, "w", encoding="utf-8") as f:
yaml.dump(data, f, default_flow_style=False, allow_unicode=True, sort_keys=False)
logger.debug(f"EARegistry: saved {len(self._profiles)} profile(s)")
def verify_integrity(self) -> list[str]:
"""
Check that all registered EAs have accessible .set files.
Returns a list of error messages (empty = all OK).
"""
errors = []
for name, profile in self._profiles.items():
if not profile.set_template_path.exists():
errors.append(f"{name}: .set file missing at {profile.set_template}")
return errors
+231
View File
@@ -0,0 +1,231 @@
"""
ea/schema.py
ParameterDef and ParameterSchema — the universal parameter representation.
Replaces mutation/param_manifest.yaml entirely.
"""
from __future__ import annotations
import math
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Optional
# ── Parameter types ───────────────────────────────────────────────────────────
PARAM_TYPES = {"float", "int", "bool", "enum", "fixed"}
@dataclass
class ParameterDef:
"""
One EA input parameter, as parsed from a .set file.
type:
float — continuous, has decimal step
int — discrete integer range
bool — true/false toggle (min=0, max=1, step=1)
enum — small discrete set of integer values
fixed — never changed during optimization (min==max, or zeroed range)
"""
name: str
default: Any # Value at default (typed: float/int/bool)
type: str # float | int | bool | enum | fixed
min: Optional[float] = None
max: Optional[float] = None
step: Optional[float] = None
optimize: bool = False # User has selected this for optimization
enum_values: list = field(default_factory=list) # Populated for type==enum
def __post_init__(self):
assert self.type in PARAM_TYPES, f"Unknown param type '{self.type}' for {self.name}"
if self.type == "enum" and not self.enum_values and self.min is not None:
# Auto-populate enum values from range
v = self.min
while v <= self.max + 1e-9:
self.enum_values.append(int(round(v)))
v += self.step
@property
def range_label(self) -> str:
"""Human-readable range string, e.g. '0.5 3.0 (step 0.5)'."""
if self.type == "fixed":
return f"{self.default} (fixed)"
if self.type == "bool":
return "true / false"
if self.type == "enum":
return " | ".join(str(v) for v in self.enum_values)
return f"{self.min} {self.max} (step {self.step})"
def clamp(self, value: float) -> Any:
"""Clamp a proposed value to valid range, return correctly typed value."""
if self.type == "fixed":
return self.default
if self.type == "bool":
return bool(round(value))
if self.min is not None:
value = max(float(self.min), min(float(self.max), float(value)))
if self.type == "int":
return int(round(value))
if self.type == "enum":
# Snap to nearest enum value
return min(self.enum_values, key=lambda v: abs(v - value))
# float — round to same decimal places as step
if self.step and self.step > 0:
decimals = len(str(self.step).rstrip("0").split(".")[-1]) if "." in str(self.step) else 0
return round(value, decimals)
return value
def step_up(self, current: Any) -> Optional[Any]:
"""Return value one step above current, or None if already at max."""
if self.type in ("fixed", "bool"):
return None
if self.type == "enum":
idx = self.enum_values.index(int(current)) if int(current) in self.enum_values else -1
return self.enum_values[idx + 1] if idx < len(self.enum_values) - 1 else None
nxt = float(current) + float(self.step)
return self.clamp(nxt) if nxt <= self.max + 1e-9 else None
def step_down(self, current: Any) -> Optional[Any]:
"""Return value one step below current, or None if already at min."""
if self.type in ("fixed", "bool"):
return None
if self.type == "enum":
idx = self.enum_values.index(int(current)) if int(current) in self.enum_values else -1
return self.enum_values[idx - 1] if idx > 0 else None
nxt = float(current) - float(self.step)
return self.clamp(nxt) if nxt >= self.min - 1e-9 else None
# ── ParameterSchema ───────────────────────────────────────────────────────────
class ParameterSchema:
"""
Full parameter schema for one EA, derived from its .set file.
Replaces mutation/param_manifest.yaml.
"""
def __init__(self, ea_name: str, source_set: Path, parameters: dict[str, ParameterDef]):
self.ea_name = ea_name
self.source_set = source_set
self.parameters = parameters # ordered dict: name → ParameterDef
# ── Accessors ─────────────────────────────────────────────────────────────
def optimizable(self) -> list[ParameterDef]:
"""Parameters the user has selected to optimize."""
return [p for p in self.parameters.values() if p.optimize]
def fixed(self) -> list[ParameterDef]:
"""Parameters that never change."""
return [p for p in self.parameters.values() if not p.optimize]
def all_params(self) -> list[ParameterDef]:
return list(self.parameters.values())
def get(self, name: str) -> Optional[ParameterDef]:
return self.parameters.get(name)
def __len__(self) -> int:
return len(self.parameters)
# ── Baseline / override helpers ───────────────────────────────────────────
def defaults(self) -> dict[str, Any]:
"""All parameters at their default values."""
return {name: p.default for name, p in self.parameters.items()}
def with_overrides(self, overrides: dict[str, Any]) -> dict[str, Any]:
"""
Merge override values with defaults.
Overrides only apply to known parameters; unknown keys are dropped.
Values are clamped to valid range.
"""
result = self.defaults()
for name, value in overrides.items():
if name in self.parameters:
result[name] = self.parameters[name].clamp(value)
return result
# ── INI rendering ─────────────────────────────────────────────────────────
def to_ini_inputs(self, params: dict[str, Any], optimize_mode: bool = False) -> str:
"""
Render [TesterInputs] block for an MT5 .ini file.
optimize_mode=False → plain values (Phase B single backtest)
optimize_mode=True → value|min|max|step ranges (Phase A genetic search)
Bool optimization ranges use 1/0 (not true/false) per MT5 spec.
"""
lines = []
full = self.with_overrides(params)
for name, p in self.parameters.items():
value = full.get(name, p.default)
if optimize_mode and p.optimize and p.type != "fixed":
# Write optimization range — booleans use 1/0 in range format
if p.type == "bool":
v = "1" if value else "0"
lines.append(f"{name}={v}|0|1|1")
else:
formatted = self._fmt(p, value)
lines.append(
f"{name}={formatted}|{self._fmt(p, p.min)}|{self._fmt(p, p.max)}|{self._fmt(p, p.step)}"
)
else:
lines.append(f"{name}={self._fmt(p, value)}")
return "\n".join(lines)
def to_set_file(self, params: dict[str, Any], header_comment: str = "") -> str:
"""
Render a clean output .set file (no optimization ranges).
This is what gets downloaded by the user.
"""
lines = []
if header_comment:
for line in header_comment.strip().splitlines():
lines.append(f"; {line}")
lines.append("")
full = self.with_overrides(params)
for name, p in self.parameters.items():
value = full.get(name, p.default)
lines.append(f"{name}={self._fmt(p, value)}")
return "\n".join(lines)
# ── Internal formatting ───────────────────────────────────────────────────
@staticmethod
def _fmt(p: ParameterDef, value: Any) -> str:
"""Format a value according to parameter type."""
if value is None:
return str(p.default)
if p.type == "bool":
# Accept int (0/1) or bool
if isinstance(value, str):
return value.lower()
return "true" if value else "false"
if p.type in ("int", "enum"):
return str(int(round(float(value))))
if p.type == "float":
step = p.step or 0.1
decimals = 0
if "." in str(step):
decimals = len(str(step).rstrip("0").split(".")[-1])
return f"{float(value):.{decimals}f}"
# fixed or unknown
return str(value)
# ── Summary ───────────────────────────────────────────────────────────────
def summary(self) -> str:
opt = self.optimizable()
return (
f"ParameterSchema({self.ea_name}): "
f"{len(self.parameters)} params total, "
f"{len(opt)} optimizable"
)
+278
View File
@@ -0,0 +1,278 @@
"""
ea/set_parser.py
Parse any MT5 .set file into a ParameterSchema.
Handles both .set formats:
value|min|max|step (single pipe — most common)
value||min||max||step||Y/N (double pipe — some MT5 builds)
Fixed detection:
min == max → type="fixed" (e.g. InpMagicNumber=202402|202402|202402|1)
min == 0 AND max == 0 → type="fixed" (zeroed range = "don't optimize")
No range at all → type="fixed"
Type detection (non-fixed only):
min==0, max==1, step==1 → bool
"." in step string → float
max - min <= 8, step==1 → enum (small discrete integer set)
otherwise → int
"""
from __future__ import annotations
import re
from pathlib import Path
from typing import Any, Optional
from loguru import logger
from ea.schema import ParameterDef, ParameterSchema
# ── Tester section keys to skip (not EA inputs) ───────────────────────────────
_TESTER_KEYS = {
"expert", "symbol", "period", "optimization", "model",
"fromdate", "todate", "forwardmode", "report", "replacereport",
"shutdownterminal", "deposit", "currency", "leverage",
"optimizationmode", "forwarddate", "optimizationiterations",
}
# EA params that should always be fixed even if they have a range
_FORCE_FIXED_PATTERNS = [
"testermode", "testeri", "testerinit", "showpanel",
"magicnumber", "magic",
]
class SetParser:
"""
Parses a MT5 .set file into a ParameterSchema.
Usage:
parser = SetParser()
schema = parser.parse(
path=Path("C:/MT5 Set files/LEGSTECH_EA_V2.set"),
ea_name="LEGSTECH_EA_V2",
default_optimize=False, # user chooses via UI
)
"""
def parse(
self,
path: Path,
ea_name: str,
default_optimize: bool = False,
force_optimize: Optional[set[str]] = None,
force_fixed: Optional[set[str]] = None,
) -> ParameterSchema:
"""
Parse a .set file and return a ParameterSchema.
Args:
path: Path to the .set file.
ea_name: Display name for the EA.
default_optimize: Whether to mark all optimizable params as optimize=True by default.
If False (default), the user selects via UI.
force_optimize: Set of param names that are always optimize=True regardless.
force_fixed: Set of param names that are always type="fixed".
"""
force_optimize = force_optimize or set()
force_fixed = force_fixed or set()
path = Path(path)
if not path.exists():
raise FileNotFoundError(f".set file not found: {path}")
try:
text = path.read_text(encoding="utf-16")
except UnicodeError:
text = path.read_text(encoding="utf-8", errors="replace")
parameters: dict[str, ParameterDef] = {}
current_section = ""
for raw_line in text.splitlines():
line = raw_line.strip()
if not line or line.startswith(";"):
continue
# Section header
if line.startswith("[") and line.endswith("]"):
current_section = line[1:-1].lower()
continue
# Skip lines without "="
if "=" not in line:
continue
name, _, rest = line.partition("=")
name = name.strip()
rest = rest.strip()
# Skip tester-section metadata keys
if name.lower() in _TESTER_KEYS:
continue
# Parse the value + optional range
param = self._parse_param(name, rest)
if param is None:
logger.debug(f"SetParser: skipped unrecognised line: {line!r}")
continue
# Apply force-fixed overrides
if name in force_fixed or self._is_force_fixed(name):
param.type = "fixed"
param.optimize = False
elif name in force_optimize:
param.optimize = True
elif default_optimize and param.type != "fixed":
param.optimize = True
parameters[name] = param
if not parameters:
raise ValueError(f"No EA input parameters found in .set file: {path}")
schema = ParameterSchema(ea_name=ea_name, source_set=path, parameters=parameters)
logger.info(f"SetParser: parsed {schema.summary()} from {path.name}")
return schema
# ── Internal ─────────────────────────────────────────────────────────────
def _parse_param(self, name: str, rest: str) -> Optional[ParameterDef]:
"""
Parse a single parameter line.
rest is everything after the first "=" on the line.
"""
# Normalise: double-pipe "||" → single "|"
rest = re.sub(r"\|\|", "|", rest)
# Strip trailing Y/N optimize flag if present
yn_match = re.search(r"\|([YN])$", rest, re.IGNORECASE)
if yn_match:
rest = rest[:yn_match.start()]
parts = [p.strip() for p in rest.split("|")]
if len(parts) == 1:
# No range info → fixed
value = self._cast_value(parts[0])
return ParameterDef(
name=name, default=value, type="fixed",
min=None, max=None, step=None, optimize=False,
)
if len(parts) < 4:
# Incomplete range — treat as fixed
value = self._cast_value(parts[0])
return ParameterDef(
name=name, default=value, type="fixed",
min=None, max=None, step=None, optimize=False,
)
raw_val, raw_min, raw_max, raw_step = parts[0], parts[1], parts[2], parts[3]
try:
default_f = float(raw_val)
min_f = float(raw_min)
max_f = float(raw_max)
step_f = float(raw_step)
except ValueError:
value = self._cast_value(raw_val)
return ParameterDef(
name=name, default=value, type="fixed",
min=None, max=None, step=None, optimize=False,
)
# Fixed detection
is_fixed = (
abs(min_f - max_f) < 1e-9 # min == max
or (abs(min_f) < 1e-9 and abs(max_f) < 1e-9) # both zero (zeroed range)
)
if is_fixed:
return ParameterDef(
name=name,
default=self._typed_default(raw_val, raw_step),
type="fixed",
min=min_f, max=max_f, step=step_f,
optimize=False,
)
# Type detection
ptype = self._detect_type(min_f, max_f, step_f, raw_step, raw_val)
default = self._typed_cast(ptype, default_f, raw_val)
return ParameterDef(
name=name,
default=default,
type=ptype,
min=min_f,
max=max_f,
step=step_f,
optimize=False, # user sets this via UI; can be overridden by caller
)
@staticmethod
def _detect_type(min_f: float, max_f: float, step_f: float,
raw_step: str, raw_val: str) -> str:
"""Infer parameter type from its range."""
# Bool: exactly 01 with step 1
if abs(min_f) < 1e-9 and abs(max_f - 1.0) < 1e-9 and abs(step_f - 1.0) < 1e-9:
return "bool"
# Float: step has decimal component
if "." in raw_step and not raw_step.endswith(".0") and float(raw_step) % 1 != 0:
return "float"
# Also float if default value has meaningful decimal
if "." in raw_val and float(raw_val) % 1 != 0:
return "float"
# Enum: small integer set (≤ 8 distinct values, step 1)
n_values = int(round((max_f - min_f) / step_f)) + 1 if step_f > 0 else 1
if abs(step_f - 1.0) < 1e-9 and n_values <= 8:
return "enum"
return "int"
@staticmethod
def _typed_cast(ptype: str, value_f: float, raw: str) -> Any:
if ptype == "bool":
return value_f != 0 or raw.lower() in ("true", "1")
if ptype == "int":
return int(round(value_f))
if ptype == "enum":
return int(round(value_f))
return value_f # float
@staticmethod
def _typed_default(raw: str, raw_step: str) -> Any:
"""Cast a fixed-param value without range context."""
lower = raw.lower()
if lower in ("true", "false"):
return lower == "true"
try:
f = float(raw)
# Return int if it's a whole number and step is integer-like
if "." not in raw_step or raw_step.endswith(".0"):
if f == int(f):
return int(f)
return f
except ValueError:
return raw
@staticmethod
def _cast_value(raw: str) -> Any:
lower = raw.lower()
if lower in ("true", "false"):
return lower == "true"
try:
f = float(raw)
return int(f) if f == int(f) and "." not in raw else f
except ValueError:
return raw
@staticmethod
def _is_force_fixed(name: str) -> bool:
"""Return True for params that are always fixed regardless of their range."""
lower = name.lower()
return any(pat in lower for pat in _FORCE_FIXED_PATTERNS)