""" Strategy Execution Engine. Interprets structured strategy configurations, applies entry/exit rules, position sizing, and generates trade signals for the backtester. Strategy configs are JSON-serialized dicts with: - universe: target tickers - entry_rules: conditions to open positions - exit_rules: conditions to close positions - signals: which signal types to use - position_sizing: method and constraints - rebalance_frequency: how often to rebalance - constraints: sector limits, position limits - risk_management: stop loss, take profit """ from __future__ import annotations import json import logging from datetime import date from typing import Any, Dict, List, Optional import numpy as np import pandas as pd from app.services.data_ingestion.yahoo import yahoo_adapter from app.services.feature_engineering.pipeline import feature_pipeline from app.services.signals.engine import signal_engine logger = logging.getLogger(__name__) class StrategyEngine: """Execute strategies against historical data to produce position targets.""" async def evaluate_strategy( self, config: Dict[str, Any], as_of_date: Optional[date] = None, period: str = "1y", ) -> Dict[str, Any]: """ Evaluate a strategy config and produce position recommendations. Returns: Dict with target_positions, signals_used, and evaluation metadata. """ universe = config.get("universe", []) if not universe: return {"target_positions": {}, "signals": [], "error": "Empty universe"} signal_types = config.get("signals", ["momentum", "mean_reversion", "volatility"]) entry_rules = config.get("entry_rules", []) exit_rules = config.get("exit_rules", []) position_sizing = config.get("position_sizing", {"method": "equal_weight"}) constraints = config.get("constraints", {}) risk_mgmt = config.get("risk_management", {}) # 1. Generate signals for universe all_signals = await signal_engine.generate_signals(universe, signal_types, period) # 2. Evaluate entry/exit rules for each ticker position_targets: Dict[str, Dict[str, Any]] = {} for ticker in universe: ticker_signals = all_signals.get(ticker, []) if not ticker_signals: continue # Compute composite score from signals composite_score = self._compute_composite_score(ticker_signals) # Apply entry rules should_enter = self._evaluate_rules(entry_rules, composite_score, ticker_signals) should_exit = self._evaluate_rules(exit_rules, composite_score, ticker_signals) if should_enter and not should_exit: direction = "long" if composite_score > 0 else "short" position_targets[ticker] = { "direction": direction, "score": round(composite_score, 4), "strength": round(min(abs(composite_score), 1.0), 4), "signals_count": len(ticker_signals), } # 3. Apply position sizing sized_positions = self._apply_position_sizing( position_targets, position_sizing, constraints ) return { "target_positions": sized_positions, "signals_used": { ticker: sigs for ticker, sigs in all_signals.items() if ticker in sized_positions }, "universe": universe, "evaluation_date": str(as_of_date or date.today()), "total_signals": sum(len(s) for s in all_signals.values()), } def _compute_composite_score(self, signals: List[Dict[str, Any]]) -> float: """Compute weighted composite score from multiple signals.""" if not signals: return 0.0 total_score = 0.0 total_weight = 0.0 for sig in signals: value = sig.get("value", 0) strength = sig.get("strength", 0.5) direction = sig.get("direction", "neutral") # Normalize to [-1, 1] if direction == "long": dir_multiplier = 1.0 elif direction == "short": dir_multiplier = -1.0 else: dir_multiplier = 0.0 score = dir_multiplier * strength total_score += score total_weight += 1.0 return total_score / total_weight if total_weight > 0 else 0.0 def _evaluate_rules( self, rules: List[Dict[str, Any]], composite_score: float, signals: List[Dict[str, Any]], ) -> bool: """Evaluate entry or exit rules against current signals.""" if not rules: # Default: enter if composite score is significant return abs(composite_score) > 0.2 for rule in rules: rule_type = rule.get("type", "threshold") if rule_type == "threshold": threshold = rule.get("value", 0.2) operator = rule.get("operator", "gt") if operator == "gt" and composite_score > threshold: return True elif operator == "lt" and composite_score < -threshold: return True elif rule_type == "signal_count": min_signals = rule.get("min", 2) direction = rule.get("direction", "long") count = sum(1 for s in signals if s.get("direction") == direction) if count >= min_signals: return True elif rule_type == "signal_strength": min_strength = rule.get("min_strength", 0.5) strong_signals = [s for s in signals if s.get("strength", 0) >= min_strength] if len(strong_signals) >= rule.get("min_count", 1): return True return False def _apply_position_sizing( self, targets: Dict[str, Dict[str, Any]], sizing_config: Dict[str, Any], constraints: Dict[str, Any], ) -> Dict[str, Dict[str, Any]]: """Apply position sizing rules and constraints.""" if not targets: return {} method = sizing_config.get("method", "equal_weight") max_position = sizing_config.get("max_position_pct", 0.1) min_positions = constraints.get("min_positions", 1) max_sector_exposure = constraints.get("max_sector_exposure", 0.3) n_positions = max(len(targets), min_positions) if method == "equal_weight": weight = min(1.0 / n_positions, max_position) for ticker in targets: targets[ticker]["weight"] = round(weight, 4) elif method == "score_weighted": scores = {t: abs(d["score"]) for t, d in targets.items()} total_score = sum(scores.values()) or 1.0 for ticker, data in targets.items(): raw_weight = scores[ticker] / total_score targets[ticker]["weight"] = round(min(raw_weight, max_position), 4) elif method == "risk_parity": # Equal risk contribution (simplified) weight = min(1.0 / n_positions, max_position) for ticker in targets: targets[ticker]["weight"] = round(weight, 4) # Normalize weights to sum to <= 1.0 total_weight = sum(d.get("weight", 0) for d in targets.values()) if total_weight > 1.0: for ticker in targets: targets[ticker]["weight"] = round( targets[ticker]["weight"] / total_weight, 4 ) return targets @staticmethod def convert_visual_graph_to_config(graph: Dict[str, Any]) -> Dict[str, Any]: """ Convert a visual strategy builder graph (nodes + edges) into a structured strategy configuration JSON. """ nodes = graph.get("nodes", []) edges = graph.get("edges", []) config: Dict[str, Any] = { "universe": [], "entry_rules": [], "exit_rules": [], "signals": [], "position_sizing": {"method": "equal_weight", "max_position_pct": 0.1}, "rebalance_frequency": "monthly", "constraints": {}, "risk_management": {}, } for node in nodes: node_type = node.get("type", "") node_config = node.get("config", {}) if node_type == "data": tickers = node_config.get("tickers", []) config["universe"].extend(tickers) elif node_type == "indicator": signal_name = node_config.get("signal_type", "momentum") config["signals"].append(signal_name) elif node_type == "factor": config["signals"].append("factor") elif node_type == "condition": rule = { "type": node_config.get("rule_type", "threshold"), "value": node_config.get("threshold", 0.2), "operator": node_config.get("operator", "gt"), } if node_config.get("is_exit"): config["exit_rules"].append(rule) else: config["entry_rules"].append(rule) elif node_type == "allocation": config["position_sizing"] = { "method": node_config.get("method", "equal_weight"), "max_position_pct": node_config.get("max_position_pct", 0.1), } elif node_type == "risk": config["risk_management"] = { "stop_loss_pct": node_config.get("stop_loss_pct", 0.05), "take_profit_pct": node_config.get("take_profit_pct", 0.15), } # Deduplicate config["universe"] = list(set(config["universe"])) config["signals"] = list(set(config["signals"])) return config strategy_engine = StrategyEngine()