File size: 8,996 Bytes
75d9b3c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 | """
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"<strong>{s_icon} {sentiment_label}</strong><br><span style='font-size:0.75rem;'>{top_news}</span>"
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"<div class='narrative-section'>"
f"<div class='narrative-heading'><i class='fa-solid fa-chart-line'></i> Technical Analysis</div>"
f"<div class='narrative-body'>"
f"Pattern: <strong>{signal_type.replace('_', ' ').title()}</strong> detected.<br>"
f"Momentum Score: <span class='narrative-metric'>{signal.get('metrics_momentum', 0):.0f}/100</span> | "
f"Volume: <span class='narrative-metric'>{signal.get('metrics_volume', 1):.1f}x</span> avg | "
f"Confidence: <span class='narrative-metric'>{signal.get('confidence', 0)*100:.0f}%</span><br>"
f"Horizon: {horizon} trading days | Regime: {signal.get('regime', 'N/A')}"
f"</div></div>"
)
# 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"<div class='narrative-section'>"
f"<div class='narrative-heading'><i class='fa-solid fa-building-columns'></i> Fundamental Quality</div>"
f"<div class='narrative-body'>"
f"Sector: <strong>{sector_str}</strong> | Score: <span class='narrative-metric'>{fund_score:.0f}/100</span><br>"
f"P/E: {pe_str} | ROE: {roe_str} | Rev Growth: {rev_growth_str} | Margin: {margin_str}<br>"
f"Analyst: <span class='narrative-tag'>{rec_str}</span>"
f"</div></div>"
)
# 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]} <span class='narrative-source'>({n.get('source', '')})</span><br>"
else:
news_lines = "No ticker-specific news found in recent feeds.<br>"
news_narrative = (
f"<div class='narrative-section'>"
f"<div class='narrative-heading'><i class='fa-solid fa-newspaper'></i> News & Sentiment</div>"
f"<div class='narrative-body'>"
f"Overall: <span class='narrative-sentiment {sentiment_class}'>{sentiment_label}</span> "
f"(score: {sentiment_score:+.2f}) | {signal.get('news_count', 0)} articles matched<br>"
f"{news_lines}"
f"</div></div>"
)
# 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"<div class='narrative-section'>"
f"<div class='narrative-heading'><i class='fa-solid fa-shield-halved'></i> Risk & Trade Plan</div>"
f"<div class='narrative-body'>"
f"Risk Status: {risk_emoji} <strong>{risk_label}</strong> | "
f"Alpha Score: <span class='narrative-metric gold'>{signal.get('alpha_score', 0):.1f}</span><br>"
f"Entry: {entry_type.upper()} @ <span class='mono'>{entry:.2f}</span> | "
f"SL: <span class='mono negative'>{sl:.2f}</span> (-{sl_pct:.1f}%) | "
f"T1: <span class='mono positive'>{t1:.2f}</span> | "
f"T2: <span class='mono positive'>{t2:.2f}</span> | "
f"T3: <span class='mono positive'>{t3:.2f}</span><br>"
f"R:R = <strong>{signal.get('risk_reward', 0):.1f}</strong> | "
f"Qty: {quantity} shares | "
f"Position: ₹{signal.get('position_value', 0):,.0f} ({signal.get('position_pct', 0):.1f}% portfolio)"
f"</div></div>"
)
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
|