feat: lint v2 — loads 71 operators from data/operators.csv, validates field IDs against registry, improved checks

alpha_factory/deterministic/lint.py

@@ -1,37 +1,34 @@
 """
-Static Lint — Layer 2: Deterministic pre-flight checks.
+Static Lint v2 — Layer 2: Deterministic pre-flight checks.
+Now loads operators from data/operators.csv and validates fields against the registry.
 No LLM. Pure Python. Catches 100% of mechanical failures.
-Mandatory and unbypassable — no agent can override this gate.
 """
 import re
 from pathlib import Path
 from ..schemas import LintResult
+from ..data.brain_fields import FIELD_INDEX
 
 
-# Load operators catalog
 def _load_operators(path: Path = Path("data/operators.csv")) -> set[str]:
-    """Load valid operator names from operators.csv."""
+    """Load valid operator names from operators.csv (71 confirmed operators)."""
     if not path.exists():
-        # Fallback: common BRAIN operators
+        # Fallback: hardcoded from the confirmed 71-operator catalog
         return {
-            "rank", "zscore", "group_rank", "group_zscore", "quantile",
-            "ts_rank", "ts_zscore", "ts_mean", "ts_std", "ts_sum",
-            "ts_min", "ts_max", "ts_argmax", "ts_argmin", "ts_skewness",
-            "ts_kurtosis", "ts_covariance", "ts_correlation", "ts_regression",
-            "ts_decay_linear", "ts_decay_exp_window", "ts_delta", "ts_delay",
-            "ts_product", "ts_moment", "ts_entropy", "ts_av_diff",
-            "ts_hump", "ts_scale", "ts_step",
-            "abs", "log", "sign", "power", "sqrt", "max", "min",
-            "if_else", "less", "greater", "equal",
-            "winsorize", "pasteurize", "truncate",
-            "vec_avg", "vec_sum", "vec_norm", "vec_count",
-            "trade_when", "ts_backfill", "filter", "mask",
-            "tail", "group_mean", "group_sum", "group_median",
-            "group_max", "group_min", "group_count", "group_neutralize",
-            "indneutralize", "market_neutralize",
-            "sigmoid", "tanh", "relu",
-            "correlation", "covariance", "regression_neut",
-            "fraction", "bucket",
+            "abs", "bucket", "correlation", "covariance", "equal", "filter",
+            "fraction", "greater", "group_count", "group_max", "group_mean",
+            "group_median", "group_min", "group_neutralize", "group_rank",
+            "group_sum", "group_zscore", "if_else", "indneutralize", "less",
+            "log", "market_neutralize", "mask", "max", "min", "pasteurize",
+            "power", "quantile", "rank", "regression_neut", "relu", "sigmoid",
+            "sign", "sqrt", "tail", "tanh", "trade_when", "truncate",
+            "ts_argmax", "ts_argmin", "ts_av_diff", "ts_backfill",
+            "ts_correlation", "ts_covariance", "ts_decay_exp_window",
+            "ts_decay_linear", "ts_delay", "ts_delta", "ts_entropy",
+            "ts_hump", "ts_kurtosis", "ts_max", "ts_mean", "ts_min",
+            "ts_moment", "ts_product", "ts_rank", "ts_regression",
+            "ts_scale", "ts_skewness", "ts_std", "ts_step", "ts_sum",
+            "ts_zscore", "vec_avg", "vec_count", "vec_norm", "vec_sum",
+            "winsorize", "zscore",
         }
     ops = set()
     with open(path) as f:
@@ -44,23 +41,28 @@ def _load_operators(path: Path = Path("data/operators.csv")) -> set[str]:
 
 ALLOWED_OPS: set[str] | None = None
 
-# Look-ahead deny patterns — these ALWAYS indicate future data leakage
+# Look-ahead deny patterns
 LOOKAHEAD_PATTERNS = [
-    r"ts_delay\([^,]+,\s*-\d",      # negative delay = future data
-    r"\bfuture_",                     # any field named future_*
-    r"\bforward_return",              # explicit forward returns
-    r"ts_backfill\([^,]+,\s*\d{3,}", # backfill > 99 days (suspicious)
+    r"ts_delay\([^,]+,\s*-\d",
+    r"\bfuture_",
+    r"\bforward_return",
+    r"ts_backfill\([^,]+,\s*\d{3,}",
 ]
 
-# Unit-safety: top-level additive operands must be normalized
+# Unit-safety wrappers
 UNIT_SAFE_WRAPPERS = {"zscore", "rank", "quantile", "group_zscore", "group_rank", "group_neutralize", "indneutralize"}
 
+# Known BRAIN field patterns (not operators)
+KNOWN_FIELD_PREFIXES = {
+    "close", "open", "high", "low", "volume", "vwap", "returns", "cap",
+    "operating_income", "enterprise_value", "equity", "ebitda", "assets",
+    "liabilities", "gross_income", "net_income", "shares", "market",
+    "implied_volatility", "short_interest", "adv", "split", "dividend",
+}
+
 
 def lint(expression: str, operators_path: Path = Path("data/operators.csv")) -> LintResult:
-    """
-    Run all deterministic pre-flight checks on a BRAIN expression.
-    Returns LintResult with pass/fail + detailed errors.
-    """
+    """Run all deterministic pre-flight checks on a BRAIN expression."""
     global ALLOWED_OPS
     if ALLOWED_OPS is None:
         ALLOWED_OPS = _load_operators(operators_path)
@@ -68,24 +70,27 @@ def lint(expression: str, operators_path: Path = Path("data/operators.csv")) ->
     errors: list[str] = []
     warnings: list[str] = []
 
-    # ─── CHECK 1: Empty or trivially short ──────────────────────
+    # CHECK 1: Empty or trivially short
     if not expression or len(expression.strip()) < 5:
         errors.append("Expression is empty or trivially short")
         return LintResult(passed=False, errors=errors, warnings=warnings)
 
-    # ─── CHECK 2: Operator validity ─────────────────────────────
+    # CHECK 2: Operator validity (against 71-operator catalog)
     found_ops = re.findall(r"\b([a-z_]+)\s*\(", expression.lower())
+    invalid_ops = []
     for op in found_ops:
         if op not in ALLOWED_OPS and op not in {"if", "and", "or", "not"}:
-            errors.append(f"Unknown operator: '{op}' — not in operators.csv")
+            invalid_ops.append(op)
+    if invalid_ops:
+        errors.append(f"Unknown operators (not in 71-op catalog): {invalid_ops}")
 
-    # ─── CHECK 3: Look-ahead detection ──────────────────────────
+    # CHECK 3: Look-ahead detection
     for pattern in LOOKAHEAD_PATTERNS:
         match = re.search(pattern, expression, re.IGNORECASE)
         if match:
             errors.append(f"Look-ahead detected: '{match.group()}' (pattern: {pattern})")
 
-    # ─── CHECK 4: Balanced parentheses ──────────────────────────
+    # CHECK 4: Balanced parentheses
     depth = 0
     for ch in expression:
         if ch == "(":
@@ -98,36 +103,43 @@ def lint(expression: str, operators_path: Path = Path("data/operators.csv")) ->
     if depth > 0:
         errors.append(f"Unbalanced parentheses: {depth} unclosed '('")
 
-    # ─── CHECK 5: Unit-safety for additive expressions ──────────
-    # Detect pattern: weight * func(...) + weight * func(...)
+    # CHECK 5: Unit-safety for additive expressions
     additive_parts = re.split(r"\s*\+\s*", expression.strip())
     if len(additive_parts) > 1:
         for part in additive_parts:
             part_clean = part.strip()
-            # Check if the part is wrapped in a unit-safe function
             first_func = re.match(r"[\d.\-]*\s*\*?\s*([a-z_]+)\s*\(", part_clean.lower())
             if first_func:
                 func_name = first_func.group(1)
                 if func_name not in UNIT_SAFE_WRAPPERS:
                     warnings.append(
-                        f"Additive operand not unit-safe (may cause 'Incompatible unit'): "
-                        f"'{part_clean[:60]}...' — wrap in zscore/rank/quantile"
+                        f"Additive operand not unit-safe: '{part_clean[:60]}...' — wrap in zscore/rank"
                     )
 
-    # ─── CHECK 6: Suspicious patterns (warnings, not errors) ────
-    # Very high decay
+    # CHECK 6: Field validation against registry
+    # Extract potential field references (words that aren't operators and aren't numbers)
+    all_words = re.findall(r"\b([a-z][a-z0-9_]+)\b", expression.lower())
+    potential_fields = [w for w in all_words if w not in ALLOWED_OPS and not w.startswith("ts_") and not w.startswith("group_")]
+    known_fields_used = [f for f in potential_fields if f in FIELD_INDEX]
+    
+    # Warn about low-coverage fields
+    for fid in known_fields_used:
+        field = FIELD_INDEX[fid]
+        if field.coverage < 0.60:
+            warnings.append(f"Field '{fid}' has low coverage ({field.coverage:.0%}) — use ts_backfill({fid}, 30)")
+
+    # CHECK 7: Decay sanity
     decay_match = re.search(r"ts_decay_linear\([^,]+,\s*(\d+)", expression)
     if decay_match and int(decay_match.group(1)) > 20:
-        warnings.append(f"Decay of {decay_match.group(1)} days is unusually high — typical range 3-15")
+        warnings.append(f"Decay of {decay_match.group(1)} days is high — typical range 3-15")
 
-    # Very long lookback
+    # CHECK 8: Very long lookback
     for match in re.finditer(r"ts_\w+\([^,]+,\s*(\d+)", expression):
         window = int(match.group(1))
         if window > 252:
-            warnings.append(f"Window of {window} days exceeds 1 year — may have low coverage for newer stocks")
+            warnings.append(f"Window of {window} days > 1 year — may reduce coverage for newer stocks")
 
-    # ─── CHECK 7: Empty field references ────────────────────────
-    # Look for common mistakes like ts_mean(, 5) or rank()
+    # CHECK 9: Empty arguments
     empty_args = re.findall(r"\b[a-z_]+\(\s*,", expression.lower())
     if empty_args:
         errors.append(f"Empty first argument detected: {empty_args}")
@@ -136,6 +148,10 @@ def lint(expression: str, operators_path: Path = Path("data/operators.csv")) ->
     if empty_func:
         errors.append(f"Function called with no arguments: {empty_func}")
 
+    # CHECK 10: Turnover guard — must have ts_decay_linear
+    if "ts_decay_linear" not in expression.lower():
+        warnings.append("No ts_decay_linear found — turnover may exceed 70% threshold")
+
     return LintResult(
         passed=len(errors) == 0,
         errors=errors,
@@ -148,3 +164,8 @@ def quick_dedup_hash(expression: str, neutralization: str, decay: int) -> str:
     import hashlib
     key = f"{expression.strip()}|{neutralization}|{decay}"
     return hashlib.sha256(key.encode()).hexdigest()[:16]
+
+
+def validate_field_exists(field_id: str) -> bool:
+    """Check if a field ID exists in the canonical registry."""
+    return field_id in FIELD_INDEX