dashboard-pnl-visualization — 仪表盘损益可视化

dashboard-pnl-visualization - 研究笔记 实验概述 项目详情 日期 2025-12-13 目标 增强交易仪表盘，添加综合的盈亏跟踪、胜率叠加和R倍位视觉化 环境 Python 3.10, matplotlib, alpaca-py, Alpaca Trading API 状态 成功 背景 交易仪表盘通常只显示总盈亏，而不分解已实现和未实现的盈亏、成本覆盖或胜率背景。这项技能记录了将利润跟踪器数据集成到仪表盘视觉化的模式。 已验证工作流程

• DashboardConfig Dataclass

from dataclasses import dataclass, field 
from typing import List 
import json 
from pathlib import Path 
@dataclass 
class DashboardConfig: 
    """运行时配置 для监控仪表盘.""" 
    symbols: List[str] = field(default_factory=lambda: ["SPY", "QQQ", "AAPL"]) 
    lookback_days: int = 30 
    interval_seconds: int = 60 
    output_dir: str = "dashboards" 
    cost_data: float = 3.33  # 每日数据成本 
    cost_software: float = 6.67  # 每日软件成本 
    risk_limit_pct: float = 10.0  # 投资组合风险限制 
    profit_log_dir: str = "logs/live" 
    save_latest: bool = True 
    @classmethod 
    def from_json(cls, path: str) -> "DashboardConfig": 
        if not path or not Path(path).exists(): 
            return cls() 
        return cls(*json.loads(Path(path).read_text())) 

• Broker Equity History Integration

def get_equity_history( 
    self, period: str = "1M",  # '1D', '1W', '1M', '3M', '1A' 
    timeframe: str = "1D",  # '1Min', '5Min', '15Min', '1H', '1D' 
) -> List[Dict[str, Any]]: 
    """返回历史权益曲线从broker.""" 
    from alpaca.trading.requests import GetPortfolioHistoryRequest 
    request = GetPortfolioHistoryRequest(period=period, timeframe=timeframe) 
    history = self._client.get_portfolio_history(request) 
    if hasattr(history, "equity") and hasattr(history, "timestamp"): 
        return [ 
            {"timestamp": ts, "value": eq} 
            for ts, eq in zip(history.timestamp, history.equity) 
        ] 
    return [] 

def load_profit_summary(log_dir: str) -> Dict: 
    """加载利润跟踪器摘要用于胜率叠加.""" 
    path = Path(log_dir) 
    summary_files = sorted(path.glob("/profit_summary.json")) 
    if not summary_files: 
        return {} 
    with summary_files[-1].open() as f: 
        summary = json.load(f) 
    trades = summary.get("trades", []) 
    pnls = [t.get("pnl", 0.0) for t in trades] 
    wins = [p for p in pnls if p > 0] 
    # 计算亏损连续 
    loss_streak = 0 
    for pnl in reversed(pnls): 
        if pnl <= 0: 
            loss_streak += 1 
        else: 
            break 
    return { 
        "win_rate": len(wins) / len(pnls) if pnls else 0.0, 
        "loss_streak": loss_streak, 
        "trade_count": len(pnls), 
    } 

• 已实现与未实现的盈亏分解

def get_pnl_breakdown(broker, cfg, profit_overlay, open_pnl=0.0) -> dict: 
    """获取盈亏分解，包括已实现和未实现的盈亏.""" 
    account = broker.get_account() 
    equity = float(account['equity']) 
    last_equity = float(account.get('last_equity', equity)) 
    daily_pnl = equity - last_equity 
    unrealized_pnl = float(open_pnl) 
    realized_pnl = daily_pnl - unrealized_pnl 
    breakdown = { 
        'gross_pnl': daily_pnl, 
        'unrealized_pnl': unrealized_pnl, 
        'realized_pnl': realized_pnl, 
        'data_cost': cfg.cost_data, 
        'software_cost': cfg.cost_software, 
    } 
    breakdown['net_pnl'] = breakdown['gross_pnl'] - cfg.cost_data - cfg.cost_software 
    # 添加胜率叠加 
    if profit_overlay: 
        breakdown['win_rate'] = profit_overlay.get('win_rate') 
        breakdown['loss_streak'] = profit_overlay.get('loss_streak') 
    return breakdown 

• R倍位显示

# R倍位颜色编码 def get_r_color(r_multiple): if r_multiple is None: return 'gray' elif r_multiple >= 2: return 'darkgreen' # 优秀 elif r_multiple >= 1: return 'green' # 良好 else: return 'orange' # 低于目标

失败尝试（关键） 尝试 原因 教训 直接使用account.equity - account.last_equity 不能分离已实现和未实现的盈亏 传递open_pnl从位置