mirror of
https://github.com/NicolasBohn/NexQuant.git
synced 2026-07-31 17:27:42 +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.
77 lines
2.4 KiB
Python
77 lines
2.4 KiB
Python
"""
|
|
Stoploss Guard Protection
|
|
|
|
Detects clusters of stoplosses and blocks trading.
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from .base import BaseProtection, ProtectionConfig, ProtectionResult, ProtectionType, ProtectionScope
|
|
|
|
|
|
@dataclass
|
|
class StoplossGuardConfig(ProtectionConfig):
|
|
"""Configuration for StoplossGuard protection."""
|
|
max_stoplosses_in_period: int = 5 # Max stoplosses allowed
|
|
stoploss_threshold_pct: float = 0.02 # What counts as stoploss (2%)
|
|
|
|
|
|
class StoplossGuardProtection(BaseProtection):
|
|
"""
|
|
Detects stoploss clusters and blocks trading.
|
|
|
|
Multiple stoplosses in short time indicates bad market conditions
|
|
or strategy no longer working.
|
|
"""
|
|
|
|
def __init__(self, config: StoplossGuardConfig):
|
|
super().__init__(config)
|
|
self.config: StoplossGuardConfig = config
|
|
|
|
@property
|
|
def scope(self) -> ProtectionScope:
|
|
return ProtectionScope.GLOBAL
|
|
|
|
def check(
|
|
self,
|
|
returns: list[float],
|
|
timestamps: list[datetime],
|
|
current_equity: float,
|
|
peak_equity: float,
|
|
**kwargs
|
|
) -> ProtectionResult:
|
|
"""Check for stoploss clusters."""
|
|
self.record_check()
|
|
|
|
if not self.config.enabled:
|
|
return ProtectionResult(
|
|
should_block=False,
|
|
reason="Protection disabled",
|
|
protection_type=ProtectionType.STOPLOSS_GUARD
|
|
)
|
|
|
|
# Count stoplosses (large losses)
|
|
stoplosses = [
|
|
r for r in returns
|
|
if r < -self.config.stoploss_threshold_pct
|
|
]
|
|
|
|
if len(stoplosses) > self.config.max_stoplosses_in_period:
|
|
severity = len(stoplosses) / self.config.max_stoplosses_in_period
|
|
|
|
result = ProtectionResult(
|
|
should_block=True,
|
|
reason=f"{len(stoplosses)} stoplosses detected (max {self.config.max_stoplosses_in_period})",
|
|
protection_type=ProtectionType.STOPLOSS_GUARD,
|
|
severity=severity
|
|
)
|
|
self.record_check(blocked=True)
|
|
return result
|
|
|
|
return ProtectionResult(
|
|
should_block=False,
|
|
reason=f"{len(stoplosses)} stoplosses (within limit)",
|
|
protection_type=ProtectionType.STOPLOSS_GUARD,
|
|
severity=len(stoplosses) / max(1, self.config.max_stoplosses_in_period)
|
|
)
|