47. Trading Signal Explainability System¶

services/signal-explanation-center/
├── src/
│   ├── main.py
│   ├── generator/
│   │   ├── explanation_generator.py
│   │   ├── summary_generator.py
│   │   └── template_engine.py
│   ├── context/
│   │   ├── market_context_fetcher.py
│   │   ├── strategy_context_fetcher.py
│   │   └── decision_context_fetcher.py
│   ├── archive/
│   │   ├── explanation_archiver.py
│   │   ├── audit_logger.py
│   │   └── performance_analyzer.py
│   ├── analysis/
│   │   ├── decision_analyzer.py
│   │   ├── success_predictor.py
│   │   └── optimization_suggester.py
│   ├── api/
│   │   ├── explanation_api.py
│   ├── config.py
│   └── requirements.txt
├── Dockerfile
└── tests/

class ExplanationGenerator:
    def generate_explanation(self, signal, market_context, strategy_context):
        structured_data = {
            "trigger_reason": {
                "primary_reason": signal.get("reason"),
                "signal_strength": signal.get("strength", 0.0),
                "threshold_breach": signal.get("threshold_breach", {})
            },
            "market_context": {
                "volatility": market_context.get("volatility", 0.0),
                "volume_profile": market_context.get("volume_profile", {}),
                "order_book_state": market_context.get("order_book_state", {}),
                "market_regime": market_context.get("market_regime", "normal")
            },
            "strategy_context": {
                "momentum_score": strategy_context.get("momentum_score", 0.0),
                "risk_level": strategy_context.get("risk_level", "medium"),
                "performance_metrics": strategy_context.get("performance_metrics", {})
            },
            "decision_chain": {
                "signal_generation": signal.get("generation_steps", []),
                "risk_validation": signal.get("risk_checks", []),
                "execution_decision": signal.get("execution_factors", [])
            }
        }

        human_summary = self.generate_human_summary(structured_data)
        confidence_score = self.calculate_confidence_score(signal, market_context, strategy_context)
        risk_assessment = self.assess_risk_level(signal, market_context, strategy_context)

        return {
            "trade_id": signal.get("trade_id"),
            "timestamp": datetime.now().isoformat(),
            "structured_data": structured_data,
            "human_summary": human_summary,
            "confidence_score": confidence_score,
            "risk_assessment": risk_assessment
        }

class MarketContextFetcher:
    def fetch_context(self, symbol, timestamp):
        market_data = self.market_data_provider.get_market_data(symbol, timestamp)

        return {
            "volatility": self.calculate_volatility(market_data),
            "volume_profile": self.analyze_volume_profile(market_data),
            "order_book_state": self.analyze_order_book(market_data),
            "market_regime": self.classify_market_regime(market_data),
            "liquidity_conditions": self.assess_liquidity(market_data),
            "technical_indicators": self.get_technical_indicators(market_data),
            "sentiment_metrics": self.get_sentiment_metrics(symbol)
        }

class StrategyContextFetcher:
    def fetch_context(self, strategy_id, timestamp):
        return {
            "performance_metrics": self.get_performance_metrics(strategy_id),
            "strategy_state": self.get_strategy_state(strategy_id),
            "signal_history": self.get_signal_history(strategy_id),
            "exposure_metrics": self.get_exposure_metrics(strategy_id),
            "risk_metrics": self.get_risk_metrics(strategy_id),
            "momentum_score": self.calculate_momentum_score(strategy_id),
            "risk_level": self.assess_strategy_risk(strategy_id)
        }

class ExplanationArchiver:
    def archive_explanation(self, trade_id, explanation):
        self.database_client.insert("trade_explanations", {
            "trade_id": trade_id,
            "explanation": explanation,
            "created_at": datetime.now().isoformat()
        })
        self.cache[trade_id] = explanation
        self.audit_logger.log_explanation(trade_id, explanation)

    def get_explanation(self, trade_id):
        if trade_id in self.cache:
            return self.cache[trade_id]

        result = self.database_client.query(
            "SELECT explanation FROM trade_explanations WHERE trade_id = %s",
            (trade_id,)
        )

        if result:
            explanation = result[0]["explanation"]
            self.cache[trade_id] = explanation
            return explanation

        return {}

@router.post("/explanation/generate")
async def generate_explanation(request: ExplanationRequest):
    explanation = explanation_generator.generate_explanation(
        request.signal,
        request.market_context or market_context_fetcher.fetch_context(...),
        request.strategy_context or strategy_context_fetcher.fetch_context(...)
    )
    explanation_archiver.archive_explanation(request.trade_id, explanation)
    return explanation

@router.get("/explanation/{trade_id}")
async def get_explanation(trade_id: str):
    explanation = explanation_archiver.get_explanation(trade_id)
    return {"trade_id": trade_id, "explanation": explanation}

@router.get("/explanation/search")
async def search_explanations(
    signal_type: Optional[str] = None,
    risk_level: Optional[str] = None,
    min_confidence: Optional[float] = None
):
    criteria = {k: v for k, v in locals().items() if v is not None}
    explanations = explanation_archiver.search_explanations(criteria)
    return {"explanations": explanations, "count": len(explanations)}

const SignalExplanationView: React.FC = () => {
  const [explanations, setExplanations] = useState<TradingExplanation[]>([]);
  const [selectedExplanation, setSelectedExplanation] = useState<TradingExplanation | null>(null);

  return (
    <div className="signal-explanation-dashboard">
      {/* Explanation Statistics */}
      <ExplanationStatisticsPanel />

      {/* Explanation Search */}
      <ExplanationSearchPanel onSearch={searchExplanations} />

      {/* Explanation List */}
      <ExplanationListPanel 
        explanations={explanations}
        onSelect={setSelectedExplanation}
      />

      {/* Detailed Explanation View */}
      {selectedExplanation && (
        <DetailedExplanationPanel explanation={selectedExplanation} />
      )}

      {/* Decision Chain Visualization */}
      {selectedExplanation && (
        <DecisionChainPanel 
          decisionChain={selectedExplanation.structured_data.decision_chain}
        />
      )}

      {/* Market Context Replay */}
      {selectedExplanation && (
        <MarketContextPanel 
          marketContext={selectedExplanation.structured_data.market_context}
        />
      )}
    </div>
  );
};