Spaces:
Sleeping
Sleeping
| """ | |
| 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 | |
| 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() | |