Upload alpha_factory/deterministic/lint.py with huggingface_hub

alpha_factory/deterministic/lint.py

@@ -0,0 +1,150 @@
+"""
+Static Lint — Layer 2: Deterministic pre-flight checks.
+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
+
+
+# Load operators catalog
+def _load_operators(path: Path = Path("data/operators.csv")) -> set[str]:
+    """Load valid operator names from operators.csv."""
+    if not path.exists():
+        # Fallback: common BRAIN operators
+        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",
+        }
+    ops = set()
+    with open(path) as f:
+        for line in f:
+            name = line.strip().split(",")[0].lower()
+            if name and name != "name":
+                ops.add(name)
+    return ops
+
+
+ALLOWED_OPS: set[str] | None = None
+
+# Look-ahead deny patterns — these ALWAYS indicate future data leakage
+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)
+]
+
+# Unit-safety: top-level additive operands must be normalized
+UNIT_SAFE_WRAPPERS = {"zscore", "rank", "quantile", "group_zscore", "group_rank", "group_neutralize", "indneutralize"}
+
+
+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.
+    """
+    global ALLOWED_OPS
+    if ALLOWED_OPS is None:
+        ALLOWED_OPS = _load_operators(operators_path)
+
+    errors: list[str] = []
+    warnings: list[str] = []
+
+    # ─── 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 ─────────────────────────────
+    found_ops = re.findall(r"\b([a-z_]+)\s*\(", expression.lower())
+    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")
+
+    # ─── 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 ──────────────────────────
+    depth = 0
+    for ch in expression:
+        if ch == "(":
+            depth += 1
+        elif ch == ")":
+            depth -= 1
+        if depth < 0:
+            errors.append("Unbalanced parentheses: extra closing ')'")
+            break
+    if depth > 0:
+        errors.append(f"Unbalanced parentheses: {depth} unclosed '('")
+
+    # ─── CHECK 5: Unit-safety for additive expressions ──────────
+    # Detect pattern: weight * func(...) + weight * func(...)
+    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"
+                    )
+
+    # ─── CHECK 6: Suspicious patterns (warnings, not errors) ────
+    # Very high decay
+    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")
+
+    # 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")
+
+    # ─── CHECK 7: Empty field references ────────────────────────
+    # Look for common mistakes like ts_mean(, 5) or rank()
+    empty_args = re.findall(r"\b[a-z_]+\(\s*,", expression.lower())
+    if empty_args:
+        errors.append(f"Empty first argument detected: {empty_args}")
+
+    empty_func = re.findall(r"\b[a-z_]+\(\s*\)", expression.lower())
+    if empty_func:
+        errors.append(f"Function called with no arguments: {empty_func}")
+
+    return LintResult(
+        passed=len(errors) == 0,
+        errors=errors,
+        warnings=warnings,
+    )
+
+
+def quick_dedup_hash(expression: str, neutralization: str, decay: int) -> str:
+    """Generate a deterministic hash for dedup against factor store."""
+    import hashlib
+    key = f"{expression.strip()}|{neutralization}|{decay}"
+    return hashlib.sha256(key.encode()).hexdigest()[:16]