"""
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