"""
Trading Signal Chart Layers
This module implements signal overlay layers for displaying buy/sell/hold signals
generated by trading strategies on charts. Integrates with the database signal model.
"""
import pandas as pd
import plotly.graph_objects as go
from typing import Dict, Any, Optional, List, Union, Tuple
from dataclasses import dataclass
from decimal import Decimal
from datetime import datetime
from ..error_handling import (
ChartErrorHandler, ChartError, ErrorSeverity,
DataValidationError, create_error_annotation, get_error_message
)
from .base import BaseLayer, LayerConfig
from utils.logger import get_logger
# Initialize logger
logger = get_logger("chart_signals")
@dataclass
class SignalLayerConfig(LayerConfig):
"""Extended configuration for signal layers"""
signal_types: List[str] = None # ['buy', 'sell', 'hold'] or subset
confidence_threshold: float = 0.0 # Minimum confidence to display (0.0-1.0)
show_confidence: bool = True # Show confidence in marker hover text
marker_size: int = 12 # Size of signal markers
show_price_labels: bool = True # Show price labels on signals
bot_id: Optional[int] = None # Filter signals by specific bot
def __post_init__(self):
super().__post_init__()
if self.signal_types is None:
self.signal_types = ['buy', 'sell'] # Default to buy/sell only
@dataclass
class TradeLayerConfig(LayerConfig):
"""Extended configuration for trade visualization layers"""
show_pnl: bool = True # Show profit/loss information
show_trade_lines: bool = True # Draw lines connecting entry/exit points
show_quantity: bool = True # Show trade quantity in hover
show_fees: bool = True # Show fees in hover
min_pnl_display: Optional[float] = None # Minimum P&L to display trade
bot_id: Optional[int] = None # Filter trades by specific bot
trade_marker_size: int = 14 # Size of trade markers (slightly larger than signals)
def __post_init__(self):
super().__post_init__()
class BaseSignalLayer(BaseLayer):
"""
Base class for all signal layers with database integration.
"""
def __init__(self, config: SignalLayerConfig):
"""
Initialize base signal layer.
Args:
config: Signal layer configuration
"""
super().__init__(config)
self.signal_data = None
# Signal styling defaults
self.signal_colors = {
'buy': '#4caf50', # Green
'sell': '#f44336', # Red
'hold': '#ff9800' # Orange
}
self.signal_symbols = {
'buy': 'triangle-up',
'sell': 'triangle-down',
'hold': 'circle'
}
def validate_signal_data(self, signals: Union[pd.DataFrame, List[Dict[str, Any]]]) -> bool:
"""
Validate signal data structure and requirements.
Args:
signals: Signal data from database or API
Returns:
True if data is valid for signal rendering
"""
try:
# Clear previous errors
self.error_handler.clear_errors()
# Convert to DataFrame if needed
if isinstance(signals, list):
if not signals:
# Empty signals are valid (no signals to show)
return True
df = pd.DataFrame(signals)
else:
df = signals.copy()
# Check required columns for signals
required_columns = ['timestamp', 'signal_type', 'price', 'confidence']
missing_columns = [col for col in required_columns if col not in df.columns]
if missing_columns:
error = ChartError(
code='MISSING_SIGNAL_COLUMNS',
message=f'Missing signal columns: {missing_columns}',
severity=ErrorSeverity.ERROR,
context={
'missing_columns': missing_columns,
'available_columns': list(df.columns),
'layer_type': 'signal'
},
recovery_suggestion=f'Ensure signal data contains: {required_columns}'
)
self.error_handler.errors.append(error)
return False
# Validate signal types
valid_signal_types = {'buy', 'sell', 'hold'}
invalid_signals = df[~df['signal_type'].isin(valid_signal_types)]
if not invalid_signals.empty:
error = ChartError(
code='INVALID_SIGNAL_TYPES',
message=f'Invalid signal types found: {set(invalid_signals["signal_type"].unique())}',
severity=ErrorSeverity.WARNING,
context={
'invalid_types': list(invalid_signals['signal_type'].unique()),
'valid_types': list(valid_signal_types)
},
recovery_suggestion='Signal types must be: buy, sell, or hold'
)
self.error_handler.warnings.append(error)
# Validate confidence range
invalid_confidence = df[(df['confidence'] < 0) | (df['confidence'] > 1)]
if not invalid_confidence.empty:
error = ChartError(
code='INVALID_CONFIDENCE_RANGE',
message=f'Confidence values must be between 0.0 and 1.0',
severity=ErrorSeverity.WARNING,
context={
'invalid_count': len(invalid_confidence),
'min_found': float(df['confidence'].min()),
'max_found': float(df['confidence'].max())
},
recovery_suggestion='Confidence values will be clamped to 0.0-1.0 range'
)
self.error_handler.warnings.append(error)
return True
except Exception as e:
self.logger.error(f"Error validating signal data: {e}")
error = ChartError(
code='SIGNAL_VALIDATION_ERROR',
message=f'Signal validation failed: {str(e)}',
severity=ErrorSeverity.ERROR,
context={'exception': str(e), 'layer_type': 'signal'}
)
self.error_handler.errors.append(error)
return False
def filter_signals_by_config(self, signals: pd.DataFrame) -> pd.DataFrame:
"""
Filter signals based on layer configuration.
Args:
signals: Raw signal data
Returns:
Filtered signal data
"""
try:
if signals.empty:
return signals
filtered = signals.copy()
# Filter by signal types
if self.config.signal_types:
filtered = filtered[filtered['signal_type'].isin(self.config.signal_types)]
# Filter by confidence threshold
if self.config.confidence_threshold > 0:
filtered = filtered[filtered['confidence'] >= self.config.confidence_threshold]
# Filter by bot_id if specified
if self.config.bot_id is not None:
if 'bot_id' in filtered.columns:
filtered = filtered[filtered['bot_id'] == self.config.bot_id]
else:
self.logger.warning(f"bot_id filter requested but no bot_id column in signal data")
# Clamp confidence values to valid range
filtered['confidence'] = filtered['confidence'].clip(0.0, 1.0)
self.logger.info(f"Filtered signals: {len(signals)} -> {len(filtered)} signals")
return filtered
except Exception as e:
self.logger.error(f"Error filtering signals: {e}")
return pd.DataFrame() # Return empty DataFrame on error
def create_signal_traces(self, signals: pd.DataFrame) -> List[go.Scatter]:
"""
Create Plotly traces for signal markers.
Args:
signals: Filtered signal data
Returns:
List of Plotly traces for each signal type
"""
traces = []
try:
if signals.empty:
return traces
# Group signals by type
for signal_type in signals['signal_type'].unique():
signal_group = signals[signals['signal_type'] == signal_type]
if signal_group.empty:
continue
# Prepare hover text
hover_text = []
for _, signal in signal_group.iterrows():
hover_parts = [
f"Signal: {signal['signal_type'].upper()}",
f"Price: ${signal['price']:.4f}",
f"Time: {signal['timestamp']}"
]
if self.config.show_confidence:
confidence_pct = signal['confidence'] * 100
hover_parts.append(f"Confidence: {confidence_pct:.1f}%")
if 'bot_id' in signal_group.columns:
hover_parts.append(f"Bot ID: {signal['bot_id']}")
hover_text.append("
".join(hover_parts))
# Create trace for this signal type
trace = go.Scatter(
x=signal_group['timestamp'],
y=signal_group['price'],
mode='markers',
marker=dict(
symbol=self.signal_symbols.get(signal_type, 'circle'),
size=self.config.marker_size,
color=self.signal_colors.get(signal_type, '#666666'),
line=dict(width=1, color='white'),
opacity=0.8
),
name=f"{signal_type.upper()} Signals",
text=hover_text,
hoverinfo='text',
showlegend=True,
legendgroup=f"signals_{signal_type}"
)
traces.append(trace)
# Add price labels if enabled
if self.config.show_price_labels:
price_trace = go.Scatter(
x=signal_group['timestamp'],
y=signal_group['price'],
mode='text',
text=[f"${price:.2f}" for price in signal_group['price']],
textposition='top center' if signal_type == 'buy' else 'bottom center',
textfont=dict(
size=8,
color=self.signal_colors.get(signal_type, '#666666')
),
showlegend=False,
hoverinfo='skip'
)
traces.append(price_trace)
return traces
except Exception as e:
self.logger.error(f"Error creating signal traces: {e}")
# Return error trace
error_trace = self.create_error_trace(f"Error displaying signals: {str(e)}")
return [error_trace]
def is_enabled(self) -> bool:
"""Check if the signal layer is enabled."""
return self.config.enabled
def is_overlay(self) -> bool:
"""Signal layers are always overlays on the main chart."""
return True
def get_subplot_row(self) -> Optional[int]:
"""Signal layers appear on main chart (no subplot)."""
return None
class BaseTradeLayer(BaseLayer):
"""
Base class for trade visualization layers with database integration.
"""
def __init__(self, config: TradeLayerConfig):
"""
Initialize base trade layer.
Args:
config: Trade layer configuration
"""
super().__init__(config)
self.trade_data = None
# Trade styling defaults
self.trade_colors = {
'buy': '#2e7d32', # Darker green for trades
'sell': '#c62828', # Darker red for trades
'profit': '#4caf50', # Green for profitable trades
'loss': '#f44336' # Red for losing trades
}
self.trade_symbols = {
'buy': 'triangle-up',
'sell': 'triangle-down'
}
def validate_trade_data(self, trades: Union[pd.DataFrame, List[Dict[str, Any]]]) -> bool:
"""
Validate trade data structure and requirements.
Args:
trades: Trade data from database
Returns:
True if data is valid for trade rendering
"""
try:
# Clear previous errors
self.error_handler.clear_errors()
# Convert to DataFrame if needed
if isinstance(trades, list):
if not trades:
# Empty trades are valid (no trades to show)
return True
df = pd.DataFrame(trades)
else:
df = trades.copy()
# Check required columns for trades
required_columns = ['timestamp', 'side', 'price', 'quantity']
missing_columns = [col for col in required_columns if col not in df.columns]
if missing_columns:
error = ChartError(
code='MISSING_TRADE_COLUMNS',
message=f'Missing trade columns: {missing_columns}',
severity=ErrorSeverity.ERROR,
context={
'missing_columns': missing_columns,
'available_columns': list(df.columns),
'layer_type': 'trade'
},
recovery_suggestion=f'Ensure trade data contains: {required_columns}'
)
self.error_handler.errors.append(error)
return False
# Validate trade sides
valid_sides = {'buy', 'sell'}
invalid_trades = df[~df['side'].isin(valid_sides)]
if not invalid_trades.empty:
error = ChartError(
code='INVALID_TRADE_SIDES',
message=f'Invalid trade sides found: {set(invalid_trades["side"].unique())}',
severity=ErrorSeverity.WARNING,
context={
'invalid_sides': list(invalid_trades['side'].unique()),
'valid_sides': list(valid_sides)
},
recovery_suggestion='Trade sides must be: buy or sell'
)
self.error_handler.warnings.append(error)
# Validate positive prices and quantities
invalid_prices = df[df['price'] <= 0]
invalid_quantities = df[df['quantity'] <= 0]
if not invalid_prices.empty:
error = ChartError(
code='INVALID_TRADE_PRICES',
message=f'Invalid trade prices found (must be > 0)',
severity=ErrorSeverity.WARNING,
context={'invalid_count': len(invalid_prices)},
recovery_suggestion='Trade prices must be positive values'
)
self.error_handler.warnings.append(error)
if not invalid_quantities.empty:
error = ChartError(
code='INVALID_TRADE_QUANTITIES',
message=f'Invalid trade quantities found (must be > 0)',
severity=ErrorSeverity.WARNING,
context={'invalid_count': len(invalid_quantities)},
recovery_suggestion='Trade quantities must be positive values'
)
self.error_handler.warnings.append(error)
return True
except Exception as e:
self.logger.error(f"Error validating trade data: {e}")
error = ChartError(
code='TRADE_VALIDATION_ERROR',
message=f'Trade validation failed: {str(e)}',
severity=ErrorSeverity.ERROR,
context={'exception': str(e), 'layer_type': 'trade'}
)
self.error_handler.errors.append(error)
return False
def filter_trades_by_config(self, trades: pd.DataFrame) -> pd.DataFrame:
"""
Filter trades based on layer configuration.
Args:
trades: Raw trade data
Returns:
Filtered trade data
"""
try:
if trades.empty:
return trades
filtered = trades.copy()
# Filter by bot_id if specified
if self.config.bot_id is not None:
if 'bot_id' in filtered.columns:
filtered = filtered[filtered['bot_id'] == self.config.bot_id]
else:
self.logger.warning(f"bot_id filter requested but no bot_id column in trade data")
# Filter by minimum P&L if specified
if self.config.min_pnl_display is not None and 'pnl' in filtered.columns:
# Only show trades with P&L above threshold (absolute value)
filtered = filtered[filtered['pnl'].abs() >= self.config.min_pnl_display]
self.logger.info(f"Filtered trades: {len(trades)} -> {len(filtered)} trades")
return filtered
except Exception as e:
self.logger.error(f"Error filtering trades: {e}")
return pd.DataFrame() # Return empty DataFrame on error
def pair_entry_exit_trades(self, trades: pd.DataFrame) -> List[Dict[str, Any]]:
"""
Pair buy and sell trades to create entry/exit connections.
Args:
trades: Filtered trade data
Returns:
List of trade pairs with entry/exit information
"""
try:
trade_pairs = []
if trades.empty:
return trade_pairs
# Sort trades by timestamp
sorted_trades = trades.sort_values('timestamp').reset_index(drop=True)
# Simple FIFO pairing logic
position = 0 # Current position (positive = long, negative = short)
open_positions = [] # Stack of open positions
for _, trade in sorted_trades.iterrows():
trade_dict = trade.to_dict()
if trade['side'] == 'buy':
# Opening long position or reducing short position
if position < 0:
# Closing short position(s)
remaining_quantity = trade['quantity']
while remaining_quantity > 0 and open_positions:
open_trade = open_positions.pop()
close_quantity = min(remaining_quantity, open_trade['quantity'])
# Create trade pair
pnl = (open_trade['price'] - trade['price']) * close_quantity
trade_pair = {
'entry_trade': open_trade,
'exit_trade': trade_dict,
'entry_time': open_trade['timestamp'],
'exit_time': trade['timestamp'],
'entry_price': open_trade['price'],
'exit_price': trade['price'],
'quantity': close_quantity,
'pnl': pnl,
'side': 'short', # This was a short position
'duration': trade['timestamp'] - open_trade['timestamp']
}
trade_pairs.append(trade_pair)
remaining_quantity -= close_quantity
open_trade['quantity'] -= close_quantity
# If open trade still has quantity, put it back
if open_trade['quantity'] > 0:
open_positions.append(open_trade)
# If there's remaining quantity, it opens a new long position
if remaining_quantity > 0:
new_trade = trade_dict.copy()
new_trade['quantity'] = remaining_quantity
open_positions.append(new_trade)
position += remaining_quantity
else:
# Opening new long position
open_positions.append(trade_dict)
position += trade['quantity']
else: # sell
# Opening short position or reducing long position
if position > 0:
# Closing long position(s)
remaining_quantity = trade['quantity']
while remaining_quantity > 0 and open_positions:
open_trade = open_positions.pop(0) # FIFO for long positions
close_quantity = min(remaining_quantity, open_trade['quantity'])
# Create trade pair
pnl = (trade['price'] - open_trade['price']) * close_quantity
trade_pair = {
'entry_trade': open_trade,
'exit_trade': trade_dict,
'entry_time': open_trade['timestamp'],
'exit_time': trade['timestamp'],
'entry_price': open_trade['price'],
'exit_price': trade['price'],
'quantity': close_quantity,
'pnl': pnl,
'side': 'long', # This was a long position
'duration': trade['timestamp'] - open_trade['timestamp']
}
trade_pairs.append(trade_pair)
remaining_quantity -= close_quantity
open_trade['quantity'] -= close_quantity
# If open trade still has quantity, put it back
if open_trade['quantity'] > 0:
open_positions.insert(0, open_trade)
# If there's remaining quantity, it opens a new short position
if remaining_quantity > 0:
new_trade = trade_dict.copy()
new_trade['quantity'] = remaining_quantity
open_positions.append(new_trade)
position -= remaining_quantity
else:
# Opening new short position
open_positions.append(trade_dict)
position -= trade['quantity']
self.logger.info(f"Paired {len(trade_pairs)} trade pairs from {len(sorted_trades)} trades")
return trade_pairs
except Exception as e:
self.logger.error(f"Error pairing trades: {e}")
return []
def is_enabled(self) -> bool:
"""Check if the trade layer is enabled."""
return self.config.enabled
def is_overlay(self) -> bool:
"""Trade layers are always overlays on the main chart."""
return True
def get_subplot_row(self) -> Optional[int]:
"""Trade layers appear on main chart (no subplot)."""
return None
class TradingSignalLayer(BaseSignalLayer):
"""
Main trading signal layer for displaying buy/sell/hold signals from database.
"""
def __init__(self, config: SignalLayerConfig = None):
"""
Initialize trading signal layer.
Args:
config: Signal layer configuration (optional, uses defaults)
"""
if config is None:
config = SignalLayerConfig(
name="Trading Signals",
enabled=True,
signal_types=['buy', 'sell'],
confidence_threshold=0.3, # Only show signals with >30% confidence
marker_size=10,
show_confidence=True,
show_price_labels=True
)
super().__init__(config)
self.logger.info(f"Initialized TradingSignalLayer: {config.name}")
def render(self, fig: go.Figure, data: pd.DataFrame, signals: pd.DataFrame = None, **kwargs) -> go.Figure:
"""
Render signal markers on the chart.
Args:
fig: Plotly figure to render onto
data: Market data (OHLCV format)
signals: Signal data from database (optional)
**kwargs: Additional rendering parameters
Returns:
Updated figure with signal overlays
"""
try:
if signals is None or signals.empty:
self.logger.info("No signals provided for rendering")
return fig
# Validate signal data
if not self.validate_signal_data(signals):
self.logger.warning("Signal data validation failed")
# Add error annotation if validation failed
error_message = self.error_handler.get_user_friendly_message()
fig.add_annotation(
text=f"Signal Error: {error_message}",
x=0.5, y=0.95,
xref="paper", yref="paper",
showarrow=False,
font=dict(color="red", size=10)
)
return fig
# Filter signals based on configuration
filtered_signals = self.filter_signals_by_config(signals)
if filtered_signals.empty:
self.logger.info("No signals remain after filtering")
return fig
# Create signal traces
signal_traces = self.create_signal_traces(filtered_signals)
# Add traces to figure
for trace in signal_traces:
fig.add_trace(trace)
# Store processed data for potential reuse
self.signal_data = filtered_signals
self.logger.info(f"Successfully rendered {len(filtered_signals)} signals")
return fig
except Exception as e:
self.logger.error(f"Error rendering signal layer: {e}")
# Add error annotation to chart
fig.add_annotation(
text=f"Signal Rendering Error: {str(e)}",
x=0.5, y=0.9,
xref="paper", yref="paper",
showarrow=False,
font=dict(color="red", size=10)
)
return fig
class TradeExecutionLayer(BaseTradeLayer):
"""
Trade execution layer for displaying actual buy/sell trades with entry/exit connections.
"""
def __init__(self, config: TradeLayerConfig = None):
"""
Initialize trade execution layer.
Args:
config: Trade layer configuration (optional, uses defaults)
"""
if config is None:
config = TradeLayerConfig(
name="Trade Executions",
enabled=True,
show_pnl=True,
show_trade_lines=True,
show_quantity=True,
show_fees=True,
trade_marker_size=12
)
super().__init__(config)
self.logger.info(f"Initialized TradeExecutionLayer: {config.name}")
def create_trade_traces(self, trades: pd.DataFrame) -> List[go.Scatter]:
"""
Create Plotly traces for trade markers and connections.
Args:
trades: Filtered trade data
Returns:
List of Plotly traces for trades
"""
traces = []
try:
if trades.empty:
return traces
# Create trade pairs for entry/exit connections
trade_pairs = self.pair_entry_exit_trades(trades)
# Create individual trade markers
for side in ['buy', 'sell']:
side_trades = trades[trades['side'] == side]
if side_trades.empty:
continue
# Prepare hover text
hover_text = []
for _, trade in side_trades.iterrows():
hover_parts = [
f"Trade: {trade['side'].upper()}",
f"Price: ${trade['price']:.4f}",
f"Time: {trade['timestamp']}"
]
if self.config.show_quantity:
hover_parts.append(f"Quantity: {trade['quantity']:.8f}")
if self.config.show_pnl and 'pnl' in trade:
pnl_value = trade.get('pnl', 0)
if pnl_value != 0:
hover_parts.append(f"P&L: ${pnl_value:.4f}")
if self.config.show_fees and 'fees' in trade:
fees = trade.get('fees', 0)
if fees > 0:
hover_parts.append(f"Fees: ${fees:.4f}")
if 'bot_id' in trade:
hover_parts.append(f"Bot ID: {trade['bot_id']}")
hover_text.append("
".join(hover_parts))
# Create trace for this trade side
trace = go.Scatter(
x=side_trades['timestamp'],
y=side_trades['price'],
mode='markers',
marker=dict(
symbol=self.trade_symbols.get(side, 'circle'),
size=self.config.trade_marker_size,
color=self.trade_colors.get(side, '#666666'),
line=dict(width=2, color='white'),
opacity=0.9
),
name=f"{side.upper()} Trades",
text=hover_text,
hoverinfo='text',
showlegend=True,
legendgroup=f"trades_{side}"
)
traces.append(trace)
# Create entry/exit connection lines if enabled
if self.config.show_trade_lines and trade_pairs:
for i, pair in enumerate(trade_pairs):
# Determine line color based on P&L
line_color = self.trade_colors['profit'] if pair['pnl'] >= 0 else self.trade_colors['loss']
# Create connection line
line_trace = go.Scatter(
x=[pair['entry_time'], pair['exit_time']],
y=[pair['entry_price'], pair['exit_price']],
mode='lines',
line=dict(
color=line_color,
width=2,
dash='solid' if pair['pnl'] >= 0 else 'dash'
),
name=f"Trade #{i+1}" if i < 10 else None, # Only show legend for first 10
showlegend=i < 10,
legendgroup=f"trade_lines",
hovertext=f"P&L: ${pair['pnl']:.4f}
Duration: {pair['duration']}",
hoverinfo='text'
)
traces.append(line_trace)
return traces
except Exception as e:
self.logger.error(f"Error creating trade traces: {e}")
# Return error trace
error_trace = self.create_error_trace(f"Error displaying trades: {str(e)}")
return [error_trace]
def render(self, fig: go.Figure, data: pd.DataFrame, trades: pd.DataFrame = None, **kwargs) -> go.Figure:
"""
Render trade execution markers and connections on the chart.
Args:
fig: Plotly figure to render onto
data: Market data (OHLCV format)
trades: Trade data from database (optional)
**kwargs: Additional rendering parameters
Returns:
Updated figure with trade overlays
"""
try:
if trades is None or trades.empty:
self.logger.info("No trades provided for rendering")
return fig
# Validate trade data
if not self.validate_trade_data(trades):
self.logger.warning("Trade data validation failed")
# Add error annotation if validation failed
error_message = self.error_handler.get_user_friendly_message()
fig.add_annotation(
text=f"Trade Error: {error_message}",
x=0.5, y=0.95,
xref="paper", yref="paper",
showarrow=False,
font=dict(color="red", size=10)
)
return fig
# Filter trades based on configuration
filtered_trades = self.filter_trades_by_config(trades)
if filtered_trades.empty:
self.logger.info("No trades remain after filtering")
return fig
# Create trade traces
trade_traces = self.create_trade_traces(filtered_trades)
# Add traces to figure
for trace in trade_traces:
fig.add_trace(trace)
# Store processed data for potential reuse
self.trade_data = filtered_trades
self.logger.info(f"Successfully rendered {len(filtered_trades)} trades")
return fig
except Exception as e:
self.logger.error(f"Error rendering trade layer: {e}")
# Add error annotation to chart
fig.add_annotation(
text=f"Trade Rendering Error: {str(e)}",
x=0.5, y=0.9,
xref="paper", yref="paper",
showarrow=False,
font=dict(color="red", size=10)
)
return fig
# Convenience functions for creating signal layers
def create_trading_signal_layer(bot_id: Optional[int] = None,
confidence_threshold: float = 0.3,
signal_types: List[str] = None,
**kwargs) -> TradingSignalLayer:
"""
Create a trading signal layer with common configurations.
Args:
bot_id: Filter signals by specific bot (None for all bots)
confidence_threshold: Minimum confidence to display signals
signal_types: Signal types to display (['buy', 'sell'] by default)
**kwargs: Additional configuration options
Returns:
Configured TradingSignalLayer instance
"""
if signal_types is None:
signal_types = ['buy', 'sell']
config = SignalLayerConfig(
name=f"Bot {bot_id} Signals" if bot_id else "Trading Signals",
enabled=True,
signal_types=signal_types,
confidence_threshold=confidence_threshold,
bot_id=bot_id,
marker_size=kwargs.get('marker_size', 10),
show_confidence=kwargs.get('show_confidence', True),
show_price_labels=kwargs.get('show_price_labels', True),
**{k: v for k, v in kwargs.items() if k not in ['marker_size', 'show_confidence', 'show_price_labels']}
)
return TradingSignalLayer(config)
def create_buy_signals_only_layer(**kwargs) -> TradingSignalLayer:
"""Create a signal layer that shows only buy signals."""
return create_trading_signal_layer(signal_types=['buy'], **kwargs)
def create_sell_signals_only_layer(**kwargs) -> TradingSignalLayer:
"""Create a signal layer that shows only sell signals."""
return create_trading_signal_layer(signal_types=['sell'], **kwargs)
def create_high_confidence_signals_layer(confidence_threshold: float = 0.7, **kwargs) -> TradingSignalLayer:
"""Create a signal layer for high-confidence signals only."""
return create_trading_signal_layer(
confidence_threshold=confidence_threshold,
**kwargs
)
# Convenience functions for creating trade layers
def create_trade_execution_layer(bot_id: Optional[int] = None,
show_pnl: bool = True,
show_trade_lines: bool = True,
**kwargs) -> TradeExecutionLayer:
"""
Create a trade execution layer with common configurations.
Args:
bot_id: Filter trades by specific bot (None for all bots)
show_pnl: Show profit/loss information
show_trade_lines: Draw lines connecting entry/exit points
**kwargs: Additional configuration options
Returns:
Configured TradeExecutionLayer instance
"""
config = TradeLayerConfig(
name=f"Bot {bot_id} Trades" if bot_id else "Trade Executions",
enabled=True,
show_pnl=show_pnl,
show_trade_lines=show_trade_lines,
bot_id=bot_id,
show_quantity=kwargs.get('show_quantity', True),
show_fees=kwargs.get('show_fees', True),
trade_marker_size=kwargs.get('trade_marker_size', 12),
min_pnl_display=kwargs.get('min_pnl_display', None),
**{k: v for k, v in kwargs.items() if k not in ['show_quantity', 'show_fees', 'trade_marker_size', 'min_pnl_display']}
)
return TradeExecutionLayer(config)
def create_profitable_trades_only_layer(**kwargs) -> TradeExecutionLayer:
"""Create a trade layer that shows only profitable trades."""
return create_trade_execution_layer(min_pnl_display=0.01, **kwargs)
def create_losing_trades_only_layer(**kwargs) -> TradeExecutionLayer:
"""Create a trade layer that shows only losing trades (for analysis)."""
config = kwargs.copy()
config['min_pnl_display'] = -float('inf') # Show all losing trades
layer = create_trade_execution_layer(**config)
# Override filter to show only losing trades
original_filter = layer.filter_trades_by_config
def losing_trades_filter(trades):
filtered = original_filter(trades)
if not filtered.empty and 'pnl' in filtered.columns:
filtered = filtered[filtered['pnl'] < 0]
return filtered
layer.filter_trades_by_config = losing_trades_filter
return layer