mirror of
https://github.com/NicolasBohn/NexQuant.git
synced 2026-07-28 07:57:44 +00:00
TPTBusiness
bd025e50dc
Implement automatic trading protection system to prevent excessive losses: PROTECTIONS (100% original code, NOT copied from Freqtrade): - Max Drawdown Protection: Blocks trading when DD > 15% (configurable) - Cooldown Period: 4h mandatory rest after 5% loss - Stoploss Guard: Detects stoploss clusters (>5 per day) - Low Performance Filter: Filters factors with Sharpe < 0.5, Win Rate < 40% ARCHITECTURE: - Base protection interface with common utilities - 4 specialized protection implementations - ProtectionManager orchestrates all active protections - Time-based blocking with automatic expiry TESTS (32 total, ALL PASS): - 25 unit tests in test/backtesting/test_protections.py - 7 integration tests in test/integration/test_all_features.py - Tests cover: normal operation, edge cases, error handling DOCUMENTATION: - Update QWEN.md with development guidelines for AI assistant * Mandatory rules: Update QWEN.md, README, requirements.txt, tests * Pre-commit checklist * Example workflow - Update README.md with protection system features - Update project structure with new modules All code is 100% original - NO license issues with Freqtrade GPLv3.
87 lines
2.9 KiB
Python
87 lines
2.9 KiB
Python
"""
|
|
Cooldown Period Protection
|
|
|
|
Enforces mandatory rest periods after losses.
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional
|
|
from .base import BaseProtection, ProtectionConfig, ProtectionResult, ProtectionType, ProtectionScope
|
|
|
|
|
|
@dataclass
|
|
class CooldownConfig(ProtectionConfig):
|
|
"""Configuration for Cooldown protection."""
|
|
cooldown_after_loss_pct: float = 0.05 # Cooldown after 5% loss
|
|
cooldown_duration_hours: int = 4 # How long to wait
|
|
|
|
|
|
class CooldownProtection(BaseProtection):
|
|
"""
|
|
Enforces cooling-off period after significant losses.
|
|
|
|
Prevents revenge trading and emotional decisions by forcing
|
|
a mandatory break after losses exceed threshold.
|
|
"""
|
|
|
|
def __init__(self, config: CooldownConfig):
|
|
super().__init__(config)
|
|
self.config: CooldownConfig = config
|
|
self.last_loss_time: Optional[datetime] = None
|
|
self.last_loss_pct: float = 0.0
|
|
|
|
@property
|
|
def scope(self) -> ProtectionScope:
|
|
return ProtectionScope.FACTOR
|
|
|
|
def check(
|
|
self,
|
|
returns: list[float],
|
|
timestamps: list[datetime],
|
|
current_equity: float,
|
|
peak_equity: float,
|
|
**kwargs
|
|
) -> ProtectionResult:
|
|
"""Check if cooldown should be triggered."""
|
|
self.record_check()
|
|
|
|
if not self.config.enabled:
|
|
return ProtectionResult(
|
|
should_block=False,
|
|
reason="Protection disabled",
|
|
protection_type=ProtectionType.COOLDOWN
|
|
)
|
|
|
|
# Check most recent return
|
|
if returns:
|
|
latest_return = returns[-1]
|
|
latest_time = timestamps[-1] if timestamps else datetime.now()
|
|
|
|
if latest_return < -self.config.cooldown_after_loss_pct:
|
|
self.last_loss_time = latest_time
|
|
self.last_loss_pct = latest_return
|
|
|
|
# If recently had big loss, enforce cooldown
|
|
if self.last_loss_time:
|
|
time_since_loss = datetime.now() - self.last_loss_time
|
|
if time_since_loss < timedelta(hours=self.config.cooldown_duration_hours):
|
|
remaining = timedelta(hours=self.config.cooldown_duration_hours) - time_since_loss
|
|
|
|
result = ProtectionResult(
|
|
should_block=True,
|
|
reason=f"Loss of {abs(self.last_loss_pct)*100:.1f}% - cooling down for {remaining.seconds // 3600}h",
|
|
until=self.last_loss_time + timedelta(hours=self.config.cooldown_duration_hours),
|
|
protection_type=ProtectionType.COOLDOWN,
|
|
severity=abs(self.last_loss_pct) / self.config.cooldown_after_loss_pct
|
|
)
|
|
self.record_check(blocked=True)
|
|
return result
|
|
|
|
return ProtectionResult(
|
|
should_block=False,
|
|
reason="No recent significant losses",
|
|
protection_type=ProtectionType.COOLDOWN,
|
|
severity=0.0
|
|
)
|