core/expression_factory.py

"""Grammar-safe Fast Expression candidate construction for controlled research runs."""

import random
from typing import Any, Dict


class ExpressionFactory:
    """Build only expressions whose operator arity and field units are known locally."""

    _LOOKBACKS = (3, 5, 7, 10)

    def generate(self, region: str, universe: str, delay: int, neutralization: str) -> Dict[str, Any]:
        fast, slow = sorted(random.sample(self._LOOKBACKS, 2))
        templates = (
            f"rank(ts_corr(returns, volume, {fast})) - rank(ts_delta(close, {slow}))",
            f"rank(ts_mean(close / vwap - 1, {fast})) - rank(ts_mean(volume / adv20 - 1, {slow}))",
            f"-rank(ts_delta(close / vwap, {fast})) + rank(ts_corr(returns, volume / adv20, {slow}))",
            f"rank(ts_rank(returns, {fast})) - rank(ts_rank(volume / adv20, {slow}))",
            f"rank(ts_delta(high - low, {fast})) - rank(ts_mean(close - open, {slow}))",
        )
        expression = random.choice(templates)
        if neutralization != "NONE":
            expression = f"group_neutralize({expression}, {neutralization.lower()})"

        return {
            "hypothesis": "Controlled price-volume cross-sectional signal with independently ranked components.",
            "expression": expression,
            "region": region,
            "universe": universe,
            "delay": delay,
            "decay": 5,
            "truncation": 0.08,
            "neutralization": neutralization,
            "language": "FASTEXPR",
        }