""" Execution Agent — Trade plan construction. 100% rule-based (zero LLM tokens). Builds complete trade plans with entry, targets, and scaling logic. """ import logging import time logger = logging.getLogger(__name__) def _build_summary_html(signal: dict, news_items: list, sentiment_label: str) -> str: s_icon = "🟢" if sentiment_label == "Bullish" else ("🔴" if sentiment_label == "Bearish" else "⚪") top_news = news_items[0].get("title", "") if news_items else signal.get("reason", "No recent news") if len(top_news) > 65: top_news = top_news[:62] + "..." return f"{s_icon} {sentiment_label}
{top_news}" def construct_trade_plan(signal: dict) -> dict: """ Build a complete, actionable trade plan from a signal. Returns a trade plan with: - Entry strategy (limit/market, scaling) - Stop loss (strict) - Multiple targets (T1/T2/T3) with scaling out logic - Time-based exit rules """ entry = signal.get("entry_price", 0) sl = signal.get("stop_loss", 0) t1 = signal.get("target_1", 0) t2 = signal.get("target_2", 0) t3 = signal.get("target_3", 0) quantity = signal.get("quantity", 0) signal_type = signal.get("signal_type", "") horizon = signal.get("horizon_days", 10) # ── Entry Strategy ── if signal_type in ("volume_breakout", "bb_squeeze_breakout"): entry_type = "market" entry_note = "Enter at market — breakout requires immediate execution" else: entry_type = "limit" entry_note = f"Place limit order at {entry:.2f} or better" # ── Scaling Strategy ── # Tranche 1: 60% at entry, Tranche 2: 40% on pullback confirmation tranche_1_qty = int(quantity * 0.6) or 1 tranche_2_qty = quantity - tranche_1_qty pullback_entry = round(entry * 0.985, 2) # 1.5% pullback # ── Exit Strategy ── # Scale out at targets: 40% at T1, 30% at T2, 30% at T3 exit_plan = [] remaining = quantity if t1 and remaining > 0: t1_qty = int(quantity * 0.4) or 1 exit_plan.append({"target": "T1", "price": t1, "quantity": t1_qty, "action": "Sell 40%"}) remaining -= t1_qty if t2 and remaining > 0: t2_qty = int(quantity * 0.3) or 1 exit_plan.append({"target": "T2", "price": t2, "quantity": min(t2_qty, remaining), "action": "Sell 30%"}) remaining -= t2_qty if t3 and remaining > 0: exit_plan.append({"target": "T3", "price": t3, "quantity": remaining, "action": "Sell remaining"}) # ── Risk Management Rules ── risk_rules = [ f"Hard stop loss at {sl:.2f} — NO exceptions", f"Trail stop to breakeven after T1 is hit", f"Time exit: Close position after {horizon} trading days if no target hit", "If market regime changes to high_volatility, tighten stop by 50%", ] # ── Build Detailed Narrative ── fund_data = signal.get("fundamental_data", {}) news_items = signal.get("news_items", []) risk_decision = signal.get("risk_decision", "approved") # Technical section tech_narrative = ( f"
" f"
Technical Analysis
" f"
" f"Pattern: {signal_type.replace('_', ' ').title()} detected.
" f"Momentum Score: {signal.get('metrics_momentum', 0):.0f}/100 | " f"Volume: {signal.get('metrics_volume', 1):.1f}x avg | " f"Confidence: {signal.get('confidence', 0)*100:.0f}%
" f"Horizon: {horizon} trading days | Regime: {signal.get('regime', 'N/A')}" f"
" ) # Fundamental section pe_str = f"{fund_data.get('pe_ratio', 0):.1f}" if fund_data.get('pe_ratio') else "N/A" roe_str = f"{fund_data.get('roe', 0)*100:.1f}%" if fund_data.get('roe') else "N/A" rev_growth_str = f"{fund_data.get('revenue_growth', 0)*100:.1f}%" if fund_data.get('revenue_growth') else "N/A" margin_str = f"{fund_data.get('profit_margin', 0)*100:.1f}%" if fund_data.get('profit_margin') else "N/A" rec_str = fund_data.get('recommendation', 'N/A') sector_str = fund_data.get('sector', 'Unknown') fund_score = signal.get('fundamental_score', signal.get('metrics_fundamental', 50)) fund_narrative = ( f"
" f"
Fundamental Quality
" f"
" f"Sector: {sector_str} | Score: {fund_score:.0f}/100
" f"P/E: {pe_str} | ROE: {roe_str} | Rev Growth: {rev_growth_str} | Margin: {margin_str}
" f"Analyst: {rec_str}" f"
" ) # News/Sentiment section sentiment_score = signal.get('sentiment_score', 0) sentiment_label = "Bullish" if sentiment_score > 0.2 else ("Bearish" if sentiment_score < -0.2 else "Neutral") sentiment_class = "positive" if sentiment_score > 0.2 else ("negative" if sentiment_score < -0.2 else "neutral") news_lines = "" if news_items: for n in news_items[:3]: s_icon = "🟢" if n.get("sentiment") == "positive" else ("🔴" if n.get("sentiment") == "negative" else "⚪") news_lines += f"{s_icon} {n.get('title', '')[:70]} ({n.get('source', '')})
" else: news_lines = "No ticker-specific news found in recent feeds.
" news_narrative = ( f"
" f"
News & Sentiment
" f"
" f"Overall: {sentiment_label} " f"(score: {sentiment_score:+.2f}) | {signal.get('news_count', 0)} articles matched
" f"{news_lines}" f"
" ) # Risk & Trade section sl_pct = round(abs(entry - sl) / entry * 100, 2) if entry else 0 risk_emoji = "✅" if risk_decision == "approved" else "⚠️" risk_label = "Approved" if risk_decision == "approved" else "Reduced Size" trade_narrative = ( f"
" f"
Risk & Trade Plan
" f"
" f"Risk Status: {risk_emoji} {risk_label} | " f"Alpha Score: {signal.get('alpha_score', 0):.1f}
" f"Entry: {entry_type.upper()} @ {entry:.2f} | " f"SL: {sl:.2f} (-{sl_pct:.1f}%) | " f"T1: {t1:.2f} | " f"T2: {t2:.2f} | " f"T3: {t3:.2f}
" f"R:R = {signal.get('risk_reward', 0):.1f} | " f"Qty: {quantity} shares | " f"Position: ₹{signal.get('position_value', 0):,.0f} ({signal.get('position_pct', 0):.1f}% portfolio)" f"
" ) full_narrative = tech_narrative + fund_narrative + news_narrative + trade_narrative trade_plan = { **signal, "trade_plan": { "entry": { "type": entry_type, "price": entry, "note": entry_note, "tranche_1": {"qty": tranche_1_qty, "price": entry, "note": "Initial entry"}, "tranche_2": {"qty": tranche_2_qty, "price": pullback_entry, "note": "Add on pullback"}, }, "stop_loss": { "price": sl, "type": "hard", "pct_from_entry": round(abs(entry - sl) / entry * 100, 2), }, "targets": exit_plan, "risk_rules": risk_rules, "horizon_days": horizon, "max_hold_date": f"{horizon} trading days from entry", }, "status": "pending", "created_at": time.time(), "agent_name": "execution_agent", "narrative": full_narrative, "reason_summary": _build_summary_html(signal, news_items, sentiment_label), } logger.info( f"Trade plan [{signal['ticker']}]: {entry_type} entry @ {entry:.2f}, " f"SL={sl:.2f}, T1={t1:.2f}, T2={t2:.2f}, qty={quantity}" ) return trade_plan def construct_all_plans(signals: list[dict]) -> list[dict]: """Build trade plans for all sized signals.""" plans = [construct_trade_plan(s) for s in signals] logger.info(f"Constructed {len(plans)} trade plans") return plans