Building a Filter Language — Your Scanner's Brain

# This works... until it doesn't
def screen_oversold(stock):
    return (
        stock.rsi_14 < 30
        and stock.volume > stock.avg_volume_20 * 1.5
        and stock.close > stock.ema_200
    )

# New strategy? Write another function.
# Tweak thresholds? Edit the code and redeploy.
# A/B test two filter sets? Copy-paste and pray.

import pandas as pd
import numpy as np

class IndicatorEngine:
    """Pre-compute all indicators for the universe."""

    def compute(self, df):
        """df: DataFrame with OHLCV columns, one row per stock (latest bar).
        Returns DataFrame with all indicator columns added."""

        # Moving Averages (pre-computed from bar history)
        # These would be computed from full history, stored as latest value
        # df['ema_20'], df['ema_50'], df['ema_200'] already present

        # RSI
        # df['rsi_14'] already computed from bar history

        # Bollinger Band Width
        df['bbw_20'] = (df['bb_upper_20'] - df['bb_lower_20']) / df['sma_20']

        # ATR as percentage of price
        df['atr_pct_14'] = df['atr_14'] / df['close']

        # Volume ratio
        df['vol_ratio_20'] = df['volume'] / df['avg_vol_20']

        # Breakout proximity
        df['pct_from_high_20'] = (df['highest_20'] - df['close']) / df['highest_20']

        return df


class FilterEvaluator:
    """Evaluate DSL expressions against stock data."""

    def evaluate(self, df, expression):
        """Evaluate a filter expression.

        expression: e.g. 'rsi_14 < 30 and vol_ratio_20 > 1.5'
        Returns: boolean Series (True for stocks that pass)
        """
        # pandas.eval handles AND/OR/comparison safely
        try:
            mask = df.eval(expression)
            return mask
        except Exception as e:
            print(f"Filter error: {e}")
            return pd.Series(False, index=df.index)

import ast
import operator

class SafeExprEvaluator:
    """Safely evaluate expressions with custom functions."""

    OPERATORS = {
        ast.Lt: operator.lt, ast.Gt: operator.gt,
        ast.LtE: operator.le, ast.GtE: operator.ge,
        ast.Eq: operator.eq, ast.NotEq: operator.ne,
        ast.Add: operator.add, ast.Sub: operator.sub,
        ast.Mult: operator.mul, ast.Div: operator.truediv,
        ast.And: lambda a, b: a & b,
        ast.Or: lambda a, b: a | b,
    }

    def __init__(self, stock_data):
        # stock_data: dict of variable_name -> value
        self.data = stock_data
        self.functions = {
            'ema': self._ema,
            'sma': self._sma,
            'rsi': self._rsi,
            'atr_pct': self._atr_pct,
            'avg_vol': self._avg_vol,
            'highest': self._highest,
            'near_breakout': self._near_breakout,
        }

    def _ema(self, period):
        return self.data.get(f'ema_{period}', 0)

    def _rsi(self, period):
        return self.data.get(f'rsi_{period}', 50)

    def _near_breakout(self, lookback, threshold):
        high_n = self.data.get(f'highest_{lookback}', 0)
        close = self.data['close']
        if high_n == 0: return False
        return (high_n - close) / high_n <= threshold

    def evaluate(self, expression):
        """Parse and evaluate a filter expression safely."""
        tree = ast.parse(expression, mode='eval')
        return self._eval_node(tree.body)

    def _eval_node(self, node):
        # Recursively evaluate AST nodes
        # ... (handles Name, Call, Compare, BoolOp, etc.)
        pass  # Full implementation ~80 lines

import json

class StrategyRegistry:
    """Load and manage filter strategies from config."""

    def __init__(self, config_path="strategies.json"):
        with open(config_path) as f:
            self.strategies = json.load(f)

    def get_filter(self, strategy_name):
        """Return the filter expression for a strategy."""
        return self.strategies[strategy_name]["filter"]

    def get_active(self, regime):
        """Return strategies active for the current regime."""
        return {
            name: strat
            for name, strat in self.strategies.items()
            if regime in strat["regimes"]
        }


# strategies.json
# {
#   "contraction": {
#     "filter": "bbw_20 < sma_bbw_20 and rsi_14 > 45 and atr_pct_14 < 0.03",
#     "regimes": ["risk_on", "neutral", "recovery"],
#     "weight": { "risk_on": 20, "neutral": 30, "risk_off": 5 }
#   },
#   "momentum": {
#     "filter": "ema_20 > ema_50 and ema_50 > ema_200 and rsi_14 > 55",
#     "regimes": ["risk_on", "recovery"],
#     "weight": { "risk_on": 45, "neutral": 20, "risk_off": 5 }
#   },
#   ...
# }

def diagnose_filter(evaluator, universe_df, expression, strategy_name):
    """Run a filter and print diagnostics."""
    mask = evaluator.evaluate(universe_df, expression)
    passed = mask.sum()
    total = len(universe_df)
    rate = passed / total * 100

    print(f"\n{'='*50}")
    print(f"Strategy: {strategy_name}")
    print(f"Expression: {expression}")
    print(f"Universe: {total} stocks")
    print(f"Passed: {passed} ({rate:.1f}%)")

    if rate < 1:
        print("[WARN] Too tight - consider relaxing thresholds")
    elif rate > 15:
        print("[WARN] Too loose - add criteria or tighten")
    else:
        print("[OK] Pass rate in healthy range")

    # Show sector distribution of passed stocks
    if passed > 0:
        sectors = universe_df[mask]['sector'].value_counts()
        print(f"\nSector distribution:")
        for sector, count in sectors.head(5).items():
            print(f"  {sector}: {count}")

    return universe_df[mask]

class ScannerPipeline:
    """Complete filter pipeline: load, compute, filter, report."""

    def __init__(self):
        self.indicators = IndicatorEngine()
        self.evaluator = FilterEvaluator()
        self.strategies = StrategyRegistry()

    def run(self, universe_df, regime):
        """Run all active strategies for the current regime."""

        # Step 1: Compute all indicators
        df = self.indicators.compute(universe_df)

        # Step 2: Get active strategies for this regime
        active = self.strategies.get_active(regime)
        print(f"Regime: {regime} -> {len(active)} active strategies")

        # Step 3: Run each filter
        all_candidates = {}
        for name, strat in active.items():
            passed = self.evaluator.evaluate(df, strat["filter"])
            candidates = df[passed].copy()
            candidates["strategy"] = name
            candidates["strategy_weight"] = strat["weight"][regime]
            all_candidates[name] = candidates
            print(f"  {name}: {len(candidates)} candidates")

        # Step 4: Merge and deduplicate
        merged = pd.concat(all_candidates.values())
        # If a stock appears in multiple strategies, keep the one
        # with the highest strategy weight (best regime fit)
        merged = merged.sort_values("strategy_weight", ascending=False)
        merged = merged.drop_duplicates(subset=["ticker"], keep="first")

        print(f"\nTotal unique candidates: {len(merged)}")
        return merged


# Usage
pipeline = ScannerPipeline()
candidates = pipeline.run(universe_df, regime="risk_on")
# -> Next step: scoring (Part 3)