Krikorios
48e60d015f
- Add advanced metrics dashboard with trade analytics - Add new trading components (EntryTypeAnalysis, MultiDayPositionTracker, NewsEventTracker, etc.) - Add strategy mode selector and trend confirmation - Add risk automation panel and slippage correlation analysis - Add daily trading plan enhancements with modal components - Add custom hooks (useApi, useLocalStorage, useAdvancedTradeMetrics) - Add broker service integration and trading API - Add test setup and vitest configuration - Include parquet data files for live market data - Add comprehensive documentation in docs/ folder
559 lines
20 KiB
Python
559 lines
20 KiB
Python
"""
|
|
Position Management Assistant API
|
|
Provides intelligent mitigation plans, exit strategies, and risk monitoring for active positions
|
|
"""
|
|
|
|
from fastapi import APIRouter, HTTPException, Query
|
|
from pydantic import BaseModel, Field
|
|
from typing import List, Optional, Dict, Literal
|
|
from datetime import datetime, timezone, timedelta
|
|
import numpy as np
|
|
|
|
router = APIRouter(prefix="/api/position-assistant", tags=["Position Assistant"])
|
|
|
|
|
|
class ActivePosition(BaseModel):
|
|
"""Current active position details"""
|
|
symbol: str = Field(default="XAU/USD")
|
|
direction: Literal["LONG", "SHORT"]
|
|
entry_price: float
|
|
quantity: float
|
|
stop_loss: float
|
|
take_profit: Optional[float] = None
|
|
entry_time: str
|
|
notes: Optional[str] = None
|
|
|
|
|
|
class MitigationStrategy(BaseModel):
|
|
"""Smart mitigation strategy for managing risk"""
|
|
strategy_name: str
|
|
priority: int # 1 = highest priority
|
|
action: str
|
|
trigger_price: float
|
|
reasoning: str
|
|
expected_benefit: str
|
|
risk_level: Literal["LOW", "MEDIUM", "HIGH"]
|
|
|
|
|
|
class PriceReversal(BaseModel):
|
|
"""Predicted price reversal levels and timing"""
|
|
level: float
|
|
probability: float # 0-1
|
|
timeframe: str # e.g., "2-4 hours", "End of day"
|
|
reasoning: str
|
|
confluences: List[str]
|
|
|
|
|
|
class PositionHealth(BaseModel):
|
|
"""Real-time position health assessment"""
|
|
status: Literal["HEALTHY", "AT_RISK", "CRITICAL", "WINNING"]
|
|
current_pnl: float
|
|
current_pnl_percent: float
|
|
distance_to_stop_loss: float
|
|
distance_to_stop_loss_percent: float
|
|
time_in_trade: str
|
|
recommendation: str
|
|
urgency: Literal["LOW", "MEDIUM", "HIGH", "URGENT"]
|
|
|
|
|
|
class PositionManagementPlan(BaseModel):
|
|
"""Complete position management plan"""
|
|
position: ActivePosition
|
|
current_price: float
|
|
health: PositionHealth
|
|
mitigation_strategies: List[MitigationStrategy]
|
|
reversal_zones: List[PriceReversal]
|
|
exit_plan: Dict
|
|
alerts: List[str]
|
|
next_actions: List[str]
|
|
|
|
|
|
def _calculate_position_health(
|
|
position: ActivePosition,
|
|
current_price: float
|
|
) -> PositionHealth:
|
|
"""Calculate real-time position health"""
|
|
|
|
# Calculate P&L
|
|
if position.direction == "SHORT":
|
|
pnl = (position.entry_price - current_price) * position.quantity
|
|
pnl_percent = ((position.entry_price - current_price) / position.entry_price) * 100
|
|
distance_to_sl = position.stop_loss - current_price
|
|
else: # LONG
|
|
pnl = (current_price - position.entry_price) * position.quantity
|
|
pnl_percent = ((current_price - position.entry_price) / position.entry_price) * 100
|
|
distance_to_sl = current_price - position.stop_loss
|
|
|
|
distance_to_sl_percent = (distance_to_sl / position.entry_price) * 100
|
|
|
|
# Calculate time in trade
|
|
entry_dt = datetime.fromisoformat(position.entry_time.replace('Z', '+00:00'))
|
|
now_dt = datetime.now(timezone.utc)
|
|
time_diff = now_dt - entry_dt
|
|
hours = time_diff.total_seconds() / 3600
|
|
|
|
if hours < 1:
|
|
time_in_trade = f"{int(time_diff.total_seconds() / 60)} minutes"
|
|
elif hours < 24:
|
|
time_in_trade = f"{hours:.1f} hours"
|
|
else:
|
|
time_in_trade = f"{hours/24:.1f} days"
|
|
|
|
# Determine status and urgency
|
|
if pnl > 0:
|
|
if pnl_percent > 2:
|
|
status = "WINNING"
|
|
urgency = "LOW"
|
|
recommendation = "Consider taking partial profits to secure gains"
|
|
else:
|
|
status = "HEALTHY"
|
|
urgency = "LOW"
|
|
recommendation = "Monitor for continuation or reversal signals"
|
|
else:
|
|
loss_percent_of_sl = abs(pnl_percent) / abs((position.stop_loss - position.entry_price) / position.entry_price * 100)
|
|
|
|
if loss_percent_of_sl > 0.8:
|
|
status = "CRITICAL"
|
|
urgency = "URGENT"
|
|
recommendation = "CLOSE POSITION NOW or implement emergency mitigation"
|
|
elif loss_percent_of_sl > 0.5:
|
|
status = "AT_RISK"
|
|
urgency = "HIGH"
|
|
recommendation = "Consider scaling out or tightening stop loss"
|
|
else:
|
|
status = "AT_RISK"
|
|
urgency = "MEDIUM"
|
|
recommendation = "Watch for reversal signals, keep stop loss in place"
|
|
|
|
return PositionHealth(
|
|
status=status,
|
|
current_pnl=round(pnl, 2),
|
|
current_pnl_percent=round(pnl_percent, 2),
|
|
distance_to_stop_loss=round(distance_to_sl, 2),
|
|
distance_to_stop_loss_percent=round(distance_to_sl_percent, 2),
|
|
time_in_trade=time_in_trade,
|
|
recommendation=recommendation,
|
|
urgency=urgency
|
|
)
|
|
|
|
|
|
def _generate_mitigation_strategies(
|
|
position: ActivePosition,
|
|
current_price: float,
|
|
health: PositionHealth
|
|
) -> List[MitigationStrategy]:
|
|
"""Generate smart mitigation strategies"""
|
|
|
|
strategies = []
|
|
|
|
if position.direction == "SHORT":
|
|
# SHORT position mitigation strategies
|
|
|
|
# Strategy 1: Partial close at break-even
|
|
strategies.append(MitigationStrategy(
|
|
strategy_name="Break-Even Exit (Partial)",
|
|
priority=1,
|
|
action=f"Close 50% of position at ${position.entry_price:.2f}",
|
|
trigger_price=position.entry_price,
|
|
reasoning="Lock in zero loss on half the position if price retraces to entry",
|
|
expected_benefit="Reduces risk by 50% while keeping upside exposure",
|
|
risk_level="LOW"
|
|
))
|
|
|
|
# Strategy 2: Scale out in profit
|
|
if current_price < position.entry_price:
|
|
target_1 = position.entry_price - (position.entry_price - current_price) * 1.5
|
|
strategies.append(MitigationStrategy(
|
|
strategy_name="Scale Out (First Target)",
|
|
priority=2,
|
|
action=f"Close 30% of position at ${target_1:.2f}",
|
|
trigger_price=target_1,
|
|
reasoning="Take partial profits at 1.5x current movement",
|
|
expected_benefit="Secure profits while maintaining exposure",
|
|
risk_level="LOW"
|
|
))
|
|
|
|
# Strategy 3: Move stop to break-even
|
|
if health.current_pnl > 0:
|
|
strategies.append(MitigationStrategy(
|
|
strategy_name="Move Stop to Break-Even",
|
|
priority=3,
|
|
action=f"Move stop loss from ${position.stop_loss:.2f} to ${position.entry_price:.2f}",
|
|
trigger_price=current_price,
|
|
reasoning="Eliminate downside risk once in profit",
|
|
expected_benefit="Cannot lose money on this trade anymore",
|
|
risk_level="LOW"
|
|
))
|
|
|
|
# Strategy 4: Emergency hedge
|
|
if health.status == "CRITICAL":
|
|
hedge_price = position.entry_price + (position.stop_loss - position.entry_price) * 0.5
|
|
strategies.append(MitigationStrategy(
|
|
strategy_name="Emergency Hedge (LONG)",
|
|
priority=1,
|
|
action=f"Open LONG position at ${current_price:.2f} (same size)",
|
|
trigger_price=current_price,
|
|
reasoning="Neutralize the position to stop bleeding while you reassess",
|
|
expected_benefit="Stop further losses immediately",
|
|
risk_level="HIGH"
|
|
))
|
|
|
|
# Strategy 5: Widen stop temporarily
|
|
if health.status == "AT_RISK" and health.urgency == "HIGH":
|
|
new_sl = position.stop_loss + (position.stop_loss - position.entry_price) * 0.3
|
|
strategies.append(MitigationStrategy(
|
|
strategy_name="Temporary Stop Widening",
|
|
priority=4,
|
|
action=f"Widen stop loss to ${new_sl:.2f} temporarily",
|
|
trigger_price=current_price,
|
|
reasoning="Give position room to breathe during volatility spike",
|
|
expected_benefit="Avoid premature stop-out if reversal is coming",
|
|
risk_level="MEDIUM"
|
|
))
|
|
|
|
else: # LONG position
|
|
# LONG position mitigation strategies (mirror of SHORT)
|
|
|
|
strategies.append(MitigationStrategy(
|
|
strategy_name="Break-Even Exit (Partial)",
|
|
priority=1,
|
|
action=f"Close 50% of position at ${position.entry_price:.2f}",
|
|
trigger_price=position.entry_price,
|
|
reasoning="Lock in zero loss on half the position if price retraces to entry",
|
|
expected_benefit="Reduces risk by 50% while keeping upside exposure",
|
|
risk_level="LOW"
|
|
))
|
|
|
|
if current_price > position.entry_price:
|
|
target_1 = position.entry_price + (current_price - position.entry_price) * 1.5
|
|
strategies.append(MitigationStrategy(
|
|
strategy_name="Scale Out (First Target)",
|
|
priority=2,
|
|
action=f"Close 30% of position at ${target_1:.2f}",
|
|
trigger_price=target_1,
|
|
reasoning="Take partial profits at 1.5x current movement",
|
|
expected_benefit="Secure profits while maintaining exposure",
|
|
risk_level="LOW"
|
|
))
|
|
|
|
if health.current_pnl > 0:
|
|
strategies.append(MitigationStrategy(
|
|
strategy_name="Move Stop to Break-Even",
|
|
priority=3,
|
|
action=f"Move stop loss from ${position.stop_loss:.2f} to ${position.entry_price:.2f}",
|
|
trigger_price=current_price,
|
|
reasoning="Eliminate downside risk once in profit",
|
|
expected_benefit="Cannot lose money on this trade anymore",
|
|
risk_level="LOW"
|
|
))
|
|
|
|
# Sort by priority
|
|
strategies.sort(key=lambda x: x.priority)
|
|
|
|
return strategies
|
|
|
|
|
|
def _predict_reversal_zones(
|
|
position: ActivePosition,
|
|
current_price: float
|
|
) -> List[PriceReversal]:
|
|
"""Predict potential reversal zones using technical analysis"""
|
|
|
|
reversals = []
|
|
|
|
if position.direction == "SHORT":
|
|
# For SHORT: Looking for price to drop (reversal down from current)
|
|
|
|
# Support level 1: 0.5 Fibonacci from entry to current
|
|
fib_50 = position.entry_price - (position.entry_price - current_price) * 0.5
|
|
if current_price > position.entry_price: # If against us
|
|
fib_50 = current_price - (current_price - position.entry_price) * 0.382
|
|
reversals.append(PriceReversal(
|
|
level=round(fib_50, 2),
|
|
probability=0.65,
|
|
timeframe="2-4 hours",
|
|
reasoning="38.2% Fibonacci retracement - common reversal zone",
|
|
confluences=["Fibonacci level", "Potential exhaustion zone"]
|
|
))
|
|
|
|
# Support level 2: Round number below entry
|
|
round_number = (int(position.entry_price / 100) * 100) - 100
|
|
if round_number < current_price:
|
|
reversals.append(PriceReversal(
|
|
level=round(round_number, 2),
|
|
probability=0.55,
|
|
timeframe="4-8 hours",
|
|
reasoning="Major round number psychological support",
|
|
confluences=["Round number", "Psychological level"]
|
|
))
|
|
|
|
# Support level 3: Previous day low (simulated)
|
|
prev_day_low = position.entry_price - (position.entry_price * 0.015) # 1.5% below entry
|
|
reversals.append(PriceReversal(
|
|
level=round(prev_day_low, 2),
|
|
probability=0.70,
|
|
timeframe="End of day",
|
|
reasoning="Estimated previous day low - strong support",
|
|
confluences=["Previous low", "Session support"]
|
|
))
|
|
|
|
else: # LONG
|
|
# For LONG: Looking for price to rise (reversal up from current)
|
|
|
|
fib_50 = position.entry_price + (current_price - position.entry_price) * 0.5
|
|
if current_price < position.entry_price: # If against us
|
|
fib_50 = current_price + (position.entry_price - current_price) * 0.382
|
|
reversals.append(PriceReversal(
|
|
level=round(fib_50, 2),
|
|
probability=0.65,
|
|
timeframe="2-4 hours",
|
|
reasoning="38.2% Fibonacci retracement - common reversal zone",
|
|
confluences=["Fibonacci level", "Potential exhaustion zone"]
|
|
))
|
|
|
|
round_number = (int(position.entry_price / 100) * 100) + 100
|
|
if round_number > current_price:
|
|
reversals.append(PriceReversal(
|
|
level=round(round_number, 2),
|
|
probability=0.55,
|
|
timeframe="4-8 hours",
|
|
reasoning="Major round number psychological resistance",
|
|
confluences=["Round number", "Psychological level"]
|
|
))
|
|
|
|
prev_day_high = position.entry_price + (position.entry_price * 0.015)
|
|
reversals.append(PriceReversal(
|
|
level=round(prev_day_high, 2),
|
|
probability=0.70,
|
|
timeframe="End of day",
|
|
reasoning="Estimated previous day high - strong resistance",
|
|
confluences=["Previous high", "Session resistance"]
|
|
))
|
|
|
|
# Sort by probability (highest first)
|
|
reversals.sort(key=lambda x: x.probability, reverse=True)
|
|
|
|
return reversals
|
|
|
|
|
|
def _create_exit_plan(
|
|
position: ActivePosition,
|
|
current_price: float,
|
|
health: PositionHealth,
|
|
reversals: List[PriceReversal]
|
|
) -> Dict:
|
|
"""Create comprehensive exit plan"""
|
|
|
|
plan = {
|
|
"immediate_action": None,
|
|
"optimal_exits": [],
|
|
"emergency_exit": None,
|
|
"time_based_exit": None
|
|
}
|
|
|
|
if health.status == "CRITICAL":
|
|
plan["immediate_action"] = {
|
|
"action": "CLOSE IMMEDIATELY",
|
|
"reason": "Position is critically at risk",
|
|
"price": current_price
|
|
}
|
|
plan["emergency_exit"] = {
|
|
"action": "Market order close if stop loss hit",
|
|
"trigger": position.stop_loss,
|
|
"loss_amount": health.current_pnl if health.current_pnl < 0 else 0
|
|
}
|
|
|
|
elif health.status == "WINNING":
|
|
# Build scaling out plan
|
|
if position.direction == "SHORT":
|
|
target_1 = current_price - (position.entry_price - current_price) * 0.5
|
|
target_2 = current_price - (position.entry_price - current_price) * 1.0
|
|
else:
|
|
target_1 = current_price + (current_price - position.entry_price) * 0.5
|
|
target_2 = current_price + (current_price - position.entry_price) * 1.0
|
|
|
|
plan["optimal_exits"] = [
|
|
{
|
|
"level": 1,
|
|
"price": round(target_1, 2),
|
|
"quantity_percent": 33,
|
|
"reason": "First profit target - secure initial gains"
|
|
},
|
|
{
|
|
"level": 2,
|
|
"price": round(target_2, 2),
|
|
"quantity_percent": 33,
|
|
"reason": "Second profit target - let winners run"
|
|
},
|
|
{
|
|
"level": 3,
|
|
"price": "Trailing stop",
|
|
"quantity_percent": 34,
|
|
"reason": "Trail remaining with break-even stop"
|
|
}
|
|
]
|
|
|
|
else: # AT_RISK or HEALTHY
|
|
# Exit at reversal zones
|
|
plan["optimal_exits"] = [
|
|
{
|
|
"level": i + 1,
|
|
"price": rev.level,
|
|
"quantity_percent": 100 if i == 0 else 50,
|
|
"reason": f"{rev.reasoning} ({int(rev.probability*100)}% probability)"
|
|
}
|
|
for i, rev in enumerate(reversals[:2])
|
|
]
|
|
|
|
# Time-based exit (end of day or session)
|
|
hours_in_trade = (datetime.now(timezone.utc) - datetime.fromisoformat(position.entry_time.replace('Z', '+00:00'))).total_seconds() / 3600
|
|
|
|
if hours_in_trade > 4 and health.status != "WINNING":
|
|
plan["time_based_exit"] = {
|
|
"time": "End of trading session",
|
|
"action": "Review and consider closing if no reversal",
|
|
"reason": "Avoid holding losing position overnight"
|
|
}
|
|
|
|
return plan
|
|
|
|
|
|
@router.post("/analyze", response_model=PositionManagementPlan)
|
|
async def analyze_position(
|
|
position: ActivePosition,
|
|
current_price: float = Query(..., description="Current market price")
|
|
) -> PositionManagementPlan:
|
|
"""
|
|
Analyze active position and provide comprehensive management plan
|
|
|
|
Example:
|
|
```
|
|
POST /api/position-assistant/analyze?current_price=4085
|
|
{
|
|
"direction": "SHORT",
|
|
"entry_price": 4070,
|
|
"quantity": 1.0,
|
|
"stop_loss": 4109,
|
|
"entry_time": "2025-11-24T10:00:00Z"
|
|
}
|
|
```
|
|
"""
|
|
|
|
try:
|
|
# Calculate position health
|
|
health = _calculate_position_health(position, current_price)
|
|
|
|
# Generate mitigation strategies
|
|
strategies = _generate_mitigation_strategies(position, current_price, health)
|
|
|
|
# Predict reversal zones
|
|
reversals = _predict_reversal_zones(position, current_price)
|
|
|
|
# Create exit plan
|
|
exit_plan = _create_exit_plan(position, current_price, health, reversals)
|
|
|
|
# Generate alerts
|
|
alerts = []
|
|
|
|
if health.status == "CRITICAL":
|
|
alerts.append("🚨 URGENT: Position at critical risk level")
|
|
alerts.append(f"⚠️ Stop loss ${abs(health.distance_to_stop_loss):.2f} away")
|
|
elif health.status == "AT_RISK" and health.urgency == "HIGH":
|
|
alerts.append(f"⚠️ Position down {abs(health.current_pnl_percent):.1f}%")
|
|
alerts.append("💡 Consider mitigation strategies")
|
|
elif health.status == "WINNING":
|
|
alerts.append(f"✅ Position up {health.current_pnl_percent:.1f}%")
|
|
alerts.append("🎯 Consider taking partial profits")
|
|
|
|
# Generate next actions
|
|
next_actions = []
|
|
|
|
if strategies:
|
|
top_strategy = strategies[0]
|
|
next_actions.append(f"📋 Primary: {top_strategy.action}")
|
|
|
|
if reversals:
|
|
top_reversal = reversals[0]
|
|
next_actions.append(f"🎯 Watch for reversal at ${top_reversal.level:.2f} ({top_reversal.timeframe})")
|
|
|
|
if exit_plan.get("immediate_action"):
|
|
next_actions.insert(0, f"🚨 {exit_plan['immediate_action']['action']}")
|
|
|
|
return PositionManagementPlan(
|
|
position=position,
|
|
current_price=current_price,
|
|
health=health,
|
|
mitigation_strategies=strategies,
|
|
reversal_zones=reversals,
|
|
exit_plan=exit_plan,
|
|
alerts=alerts,
|
|
next_actions=next_actions
|
|
)
|
|
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to analyze position: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.get("/quick-status")
|
|
async def get_quick_status(
|
|
direction: str = Query(..., description="LONG or SHORT"),
|
|
entry_price: float = Query(...),
|
|
current_price: float = Query(...),
|
|
stop_loss: float = Query(...)
|
|
) -> Dict:
|
|
"""
|
|
Quick position status check without full analysis
|
|
|
|
Example:
|
|
```
|
|
GET /api/position-assistant/quick-status?direction=SHORT&entry_price=4070¤t_price=4085&stop_loss=4109
|
|
```
|
|
"""
|
|
|
|
try:
|
|
# Quick P&L calculation
|
|
if direction.upper() == "SHORT":
|
|
pnl = entry_price - current_price
|
|
pnl_percent = ((entry_price - current_price) / entry_price) * 100
|
|
distance_to_sl = stop_loss - current_price
|
|
else:
|
|
pnl = current_price - entry_price
|
|
pnl_percent = ((current_price - entry_price) / entry_price) * 100
|
|
distance_to_sl = current_price - stop_loss
|
|
|
|
distance_to_sl_percent = (distance_to_sl / entry_price) * 100
|
|
|
|
# Quick status
|
|
if pnl > 0:
|
|
status = "✅ In Profit"
|
|
color = "green"
|
|
else:
|
|
loss_ratio = abs(distance_to_sl_percent / ((stop_loss - entry_price) / entry_price * 100))
|
|
if loss_ratio > 0.8:
|
|
status = "🚨 CRITICAL - Close to stop loss"
|
|
color = "red"
|
|
elif loss_ratio > 0.5:
|
|
status = "⚠️ AT RISK"
|
|
color = "orange"
|
|
else:
|
|
status = "📊 Monitoring"
|
|
color = "yellow"
|
|
|
|
return {
|
|
"status": status,
|
|
"color": color,
|
|
"pnl": round(pnl, 2),
|
|
"pnl_percent": round(pnl_percent, 2),
|
|
"distance_to_stop_loss": round(abs(distance_to_sl), 2),
|
|
"distance_to_stop_loss_percent": round(abs(distance_to_sl_percent), 2)
|
|
}
|
|
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to get quick status: {str(e)}"
|
|
)
|