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
414 lines
14 KiB
Python
414 lines
14 KiB
Python
"""
|
|
Live Performance Dashboard API - Real-time plan monitoring and alerts
|
|
"""
|
|
|
|
from fastapi import APIRouter, HTTPException, Depends, Query
|
|
from sqlalchemy.orm import Session
|
|
from typing import Any, Dict, List, Optional, Literal
|
|
from datetime import datetime, date, timezone
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app.db.database import get_db
|
|
from app.models.models import DailyChecklist, UserProfile
|
|
from app.services.simulation_state import load_simulation_state
|
|
|
|
router = APIRouter(prefix="/api/live-dashboard", tags=["Live Dashboard"])
|
|
|
|
|
|
class DailyPlanStatus(BaseModel):
|
|
"""Current status of today's trading plan"""
|
|
date: str
|
|
target: float
|
|
actual_pnl: float
|
|
progress_percent: float
|
|
max_loss: float
|
|
current_drawdown: float
|
|
max_trades: int
|
|
actual_trades: int
|
|
trades_remaining: int
|
|
status: Literal["on-track", "near-limit", "limit-reached", "target-met"]
|
|
alerts: List[str]
|
|
|
|
|
|
class PerformanceWidget(BaseModel):
|
|
"""Sticky dashboard widget data"""
|
|
daily_plan: DailyPlanStatus
|
|
position_summary: Dict
|
|
risk_metrics: Dict
|
|
alerts: List[Dict]
|
|
recommendations: List[str]
|
|
|
|
|
|
class AlertConfig(BaseModel):
|
|
"""Alert configuration"""
|
|
alert_type: str # trade_limit, loss_limit, target_achieved, break_recommended
|
|
enabled: bool
|
|
threshold: Optional[float] = None
|
|
message: str
|
|
|
|
|
|
# In-memory simulation state (shared with trading.py)
|
|
|
|
|
|
def _get_today_plan_from_storage() -> Optional[Dict]:
|
|
"""Get today's trading plan blueprint (defaults until persistence is added)."""
|
|
# In production, this would query the database
|
|
# For now, we'll use a default plan structure
|
|
return {
|
|
"date": date.today().isoformat(),
|
|
"daily_target": 500.0,
|
|
"max_loss": 250.0,
|
|
"max_trades": 3,
|
|
"bias": "NEUTRAL",
|
|
}
|
|
|
|
|
|
def _calculate_daily_pnl(trades: List[Dict[str, Any]], target_date: date | None = None) -> float:
|
|
"""Calculate P&L for trades executed on the target date"""
|
|
target_date = target_date or date.today()
|
|
daily_pnl = 0.0
|
|
for trade in trades:
|
|
trade_ts = trade.get("timestamp", 0)
|
|
trade_date = datetime.fromtimestamp(trade_ts, tz=timezone.utc).date()
|
|
if trade_date == target_date:
|
|
pnl = trade.get("pnl", 0.0)
|
|
if pnl:
|
|
daily_pnl += pnl
|
|
return daily_pnl
|
|
|
|
|
|
def _count_today_trades(trades: List[Dict[str, Any]], target_date: date | None = None) -> int:
|
|
"""Count trades executed on the target date"""
|
|
target_date = target_date or date.today()
|
|
count = 0
|
|
for trade in trades:
|
|
trade_ts = trade.get("timestamp", 0)
|
|
trade_date = datetime.fromtimestamp(trade_ts, tz=timezone.utc).date()
|
|
if trade_date == target_date:
|
|
count += 1
|
|
return count
|
|
|
|
|
|
def _generate_alerts(plan: Dict, actual_pnl: float, trades_count: int) -> List[str]:
|
|
"""Generate smart alerts based on plan vs actual"""
|
|
alerts = []
|
|
|
|
target = plan.get("daily_target", 500.0)
|
|
max_loss = plan.get("max_loss", 250.0)
|
|
max_trades = plan.get("max_trades", 3)
|
|
|
|
# Trade limit alerts
|
|
trades_remaining = max_trades - trades_count
|
|
if trades_remaining == 1:
|
|
alerts.append(f"⚠️ Only 1 trade remaining before daily limit")
|
|
elif trades_remaining <= 0:
|
|
alerts.append(f"🛑 Daily trade limit reached ({max_trades} trades)")
|
|
|
|
# Loss alerts
|
|
if actual_pnl < 0:
|
|
loss_percent = (abs(actual_pnl) / max_loss) * 100
|
|
if loss_percent >= 100:
|
|
alerts.append(f"🚨 Max loss limit reached (${abs(actual_pnl):.2f})")
|
|
elif loss_percent >= 80:
|
|
alerts.append(f"⚠️ Near max loss limit ({loss_percent:.0f}% of ${max_loss})")
|
|
elif loss_percent >= 50:
|
|
alerts.append(f"⚡ Drawdown at {loss_percent:.0f}% of max loss")
|
|
|
|
# Target achievement alerts
|
|
if actual_pnl > 0:
|
|
progress_percent = (actual_pnl / target) * 100
|
|
if progress_percent >= 100:
|
|
alerts.append(f"🎉 Daily target achieved! (+${actual_pnl:.2f})")
|
|
elif progress_percent >= 80:
|
|
alerts.append(f"🎯 ${target - actual_pnl:.2f} away from daily target")
|
|
|
|
# Trading duration alerts (if 2+ hours and significant losses)
|
|
if trades_count >= 2 and actual_pnl < -100:
|
|
alerts.append(f"💡 Consider taking a break. ${abs(actual_pnl):.2f} in losses after {trades_count} trades")
|
|
|
|
return alerts
|
|
|
|
|
|
def _determine_status(
|
|
actual_pnl: float,
|
|
target: float,
|
|
max_loss: float,
|
|
trades_count: int,
|
|
max_trades: int
|
|
) -> Literal["on-track", "near-limit", "limit-reached", "target-met"]:
|
|
"""Determine overall plan status"""
|
|
|
|
# Target met
|
|
if actual_pnl >= target:
|
|
return "target-met"
|
|
|
|
# Limits reached
|
|
if trades_count >= max_trades:
|
|
return "limit-reached"
|
|
|
|
if actual_pnl <= -max_loss:
|
|
return "limit-reached"
|
|
|
|
# Near limits
|
|
loss_percent = (abs(actual_pnl) / max_loss) * 100 if actual_pnl < 0 else 0
|
|
trades_percent = (trades_count / max_trades) * 100
|
|
|
|
if loss_percent >= 80 or trades_percent >= 80:
|
|
return "near-limit"
|
|
|
|
# On track
|
|
return "on-track"
|
|
|
|
|
|
@router.get("/status", response_model=DailyPlanStatus)
|
|
async def get_dashboard_status(db: Session = Depends(get_db)) -> DailyPlanStatus:
|
|
"""
|
|
Get current status of today's trading plan with real-time metrics
|
|
"""
|
|
try:
|
|
plan = _get_today_plan_from_storage()
|
|
if not plan:
|
|
raise HTTPException(status_code=404, detail="No trading plan found for today")
|
|
|
|
state = load_simulation_state(db)
|
|
trades = state.get("trades", [])
|
|
actual_pnl = _calculate_daily_pnl(trades)
|
|
trades_count = _count_today_trades(trades)
|
|
|
|
target = plan.get("daily_target", 500.0)
|
|
max_loss = plan.get("max_loss", 250.0)
|
|
max_trades = plan.get("max_trades", 3)
|
|
|
|
progress_percent = (actual_pnl / target) * 100 if target > 0 else 0
|
|
current_drawdown = abs(actual_pnl) if actual_pnl < 0 else 0
|
|
trades_remaining = max(0, max_trades - trades_count)
|
|
|
|
alerts = _generate_alerts(plan, actual_pnl, trades_count)
|
|
status = _determine_status(actual_pnl, target, max_loss, trades_count, max_trades)
|
|
|
|
return DailyPlanStatus(
|
|
date=plan["date"],
|
|
target=target,
|
|
actual_pnl=actual_pnl,
|
|
progress_percent=round(progress_percent, 1),
|
|
max_loss=max_loss,
|
|
current_drawdown=current_drawdown,
|
|
max_trades=max_trades,
|
|
actual_trades=trades_count,
|
|
trades_remaining=trades_remaining,
|
|
status=status,
|
|
alerts=alerts,
|
|
)
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to get dashboard status: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.get("/widget", response_model=PerformanceWidget)
|
|
async def get_performance_widget(db: Session = Depends(get_db)) -> PerformanceWidget:
|
|
"""
|
|
Get complete performance widget data for sticky dashboard
|
|
"""
|
|
try:
|
|
# Get daily plan status
|
|
daily_plan = await get_dashboard_status(db=db)
|
|
|
|
state = load_simulation_state(db)
|
|
position = state.get("position")
|
|
cash = float(state.get("cash", 100000.0))
|
|
position_value = 0.0
|
|
if position:
|
|
position_value = float(position.get("quantity", 0.0)) * float(position.get("avg_price", 0.0))
|
|
|
|
total_equity = cash + position_value
|
|
position_summary = {
|
|
"has_position": position is not None,
|
|
"quantity": float(position.get("quantity", 0.0)) if position else 0,
|
|
"avg_price": float(position.get("avg_price", 0.0)) if position else 0,
|
|
"cash": cash,
|
|
"total_equity": total_equity,
|
|
}
|
|
|
|
# Calculate risk metrics
|
|
initial_capital = float(state.get("initial_capital", 100000.0))
|
|
safe_equity = total_equity if total_equity != 0 else 1
|
|
total_return = ((total_equity - initial_capital) / initial_capital) * 100 if initial_capital else 0
|
|
|
|
risk_metrics = {
|
|
"total_equity": total_equity,
|
|
"total_return_percent": round(total_return, 2),
|
|
"position_size_percent": round((position_value / safe_equity * 100), 2) if position else 0,
|
|
"cash_percent": round((cash / safe_equity * 100), 2),
|
|
}
|
|
|
|
# Generate smart recommendations
|
|
recommendations = []
|
|
|
|
if daily_plan.status == "target-met":
|
|
recommendations.append("🎉 Consider closing for the day - target achieved!")
|
|
elif daily_plan.status == "limit-reached":
|
|
recommendations.append("🛑 Trading halt recommended - daily limits reached")
|
|
elif daily_plan.status == "near-limit":
|
|
if daily_plan.trades_remaining == 1:
|
|
recommendations.append("⚠️ Last trade available - make it count")
|
|
if daily_plan.current_drawdown > daily_plan.max_loss * 0.8:
|
|
recommendations.append("🔻 Consider defensive position sizing")
|
|
else:
|
|
if daily_plan.actual_pnl > daily_plan.target * 0.7:
|
|
recommendations.append("🎯 Near target - consider taking profits")
|
|
|
|
# Alert objects with metadata
|
|
alert_objects = [
|
|
{
|
|
"type": "info",
|
|
"message": alert,
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
for alert in daily_plan.alerts
|
|
]
|
|
|
|
return PerformanceWidget(
|
|
daily_plan=daily_plan,
|
|
position_summary=position_summary,
|
|
risk_metrics=risk_metrics,
|
|
alerts=alert_objects,
|
|
recommendations=recommendations,
|
|
)
|
|
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to get performance widget: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.post("/check-limits")
|
|
async def check_trading_limits(db: Session = Depends(get_db)) -> Dict:
|
|
"""
|
|
Check if trading should be halted based on plan limits
|
|
Returns: {can_trade: bool, reason: str}
|
|
"""
|
|
try:
|
|
plan = _get_today_plan_from_storage()
|
|
if not plan:
|
|
return {"can_trade": True, "reason": "No plan configured"}
|
|
|
|
state = load_simulation_state(db)
|
|
trades = state.get("trades", [])
|
|
actual_pnl = _calculate_daily_pnl(trades)
|
|
trades_count = _count_today_trades(trades)
|
|
|
|
max_loss = plan.get("max_loss", 250.0)
|
|
max_trades = plan.get("max_trades", 3)
|
|
target = plan.get("daily_target", 500.0)
|
|
|
|
if actual_pnl <= -max_loss:
|
|
return {
|
|
"can_trade": False,
|
|
"reason": f"Max loss limit reached (${abs(actual_pnl):.2f})",
|
|
"limit_type": "loss",
|
|
}
|
|
|
|
if trades_count >= max_trades:
|
|
return {
|
|
"can_trade": False,
|
|
"reason": f"Max trades limit reached ({trades_count}/{max_trades})",
|
|
"limit_type": "trades",
|
|
}
|
|
|
|
if actual_pnl >= target:
|
|
return {
|
|
"can_trade": True,
|
|
"reason": f"Target achieved (+${actual_pnl:.2f}) - consider closing for the day",
|
|
"warning": True,
|
|
}
|
|
|
|
return {
|
|
"can_trade": True,
|
|
"reason": "Within limits",
|
|
"remaining_trades": max_trades - trades_count,
|
|
"remaining_loss_buffer": max_loss + actual_pnl,
|
|
}
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to check trading limits: {str(e)}"
|
|
)
|
|
|
|
|
|
@router.get("/session-summary")
|
|
async def get_session_summary(db: Session = Depends(get_db)) -> Dict:
|
|
"""
|
|
Get end-of-day session summary with AI coaching suggestions
|
|
"""
|
|
try:
|
|
plan = _get_today_plan_from_storage()
|
|
state = load_simulation_state(db)
|
|
trades = state.get("trades", [])
|
|
actual_pnl = _calculate_daily_pnl(trades)
|
|
trades_count = _count_today_trades(trades)
|
|
|
|
if not plan:
|
|
raise HTTPException(status_code=404, detail="No trading plan found")
|
|
|
|
target = plan.get("daily_target", 500.0)
|
|
max_loss = plan.get("max_loss", 250.0)
|
|
|
|
target_achieved = actual_pnl >= target
|
|
within_limits = actual_pnl > -max_loss and trades_count <= plan.get("max_trades", 3)
|
|
|
|
today = date.today()
|
|
today_trades = [
|
|
t for t in trades
|
|
if datetime.fromtimestamp(t.get("timestamp", 0), tz=timezone.utc).date() == today
|
|
]
|
|
|
|
winning_trades = sum(1 for t in today_trades if t.get("pnl", 0) > 0)
|
|
win_rate = (winning_trades / len(today_trades) * 100) if today_trades else 0
|
|
|
|
coaching = []
|
|
if target_achieved:
|
|
coaching.append("✅ Excellent discipline - you met your daily target!")
|
|
else:
|
|
deficit = target - actual_pnl
|
|
coaching.append(f"📊 ${deficit:.2f} short of target. Review your entry setups.")
|
|
|
|
if win_rate >= 60:
|
|
coaching.append(f"🎯 Strong win rate ({win_rate:.0f}%). Keep following your strategy.")
|
|
elif win_rate < 40:
|
|
coaching.append(f"⚠️ Low win rate ({win_rate:.0f}%). Review your trade selection criteria.")
|
|
|
|
if not within_limits:
|
|
coaching.append("🔻 Limits exceeded. Focus on risk management tomorrow.")
|
|
if trades_count > plan.get("max_trades", 3):
|
|
coaching.append("⚠️ Over-trading detected. Stick to your max trades limit.")
|
|
|
|
return {
|
|
"date": plan["date"],
|
|
"summary": {
|
|
"target": target,
|
|
"actual_pnl": actual_pnl,
|
|
"target_achieved": target_achieved,
|
|
"within_limits": within_limits,
|
|
"trades_count": trades_count,
|
|
"win_rate": round(win_rate, 1),
|
|
},
|
|
"coaching": coaching,
|
|
"next_session_suggestions": [
|
|
"Review today's winning trades for patterns",
|
|
"Adjust stop loss strategy if needed",
|
|
"Focus on high-probability setups only",
|
|
],
|
|
}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=500,
|
|
detail=f"Failed to generate session summary: {str(e)}"
|
|
)
|