142 lines
4.8 KiB
Python
142 lines
4.8 KiB
Python
"""
|
|
ML-Powered Adaptive Trading Strategy für Trading Bot V2
|
|
Ersetzt die alte DCA-Strategie
|
|
"""
|
|
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional, Dict, List
|
|
from pydantic import BaseModel
|
|
import joblib
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
class MLStrategy(BaseModel):
|
|
"""ML-based trading strategy with adaptive position sizing."""
|
|
|
|
trading_pair: str = "BTCUSDT" # Oder ETH, SOL
|
|
min_prob_threshold: float = 0.60 # Only trade if prob >= 60%
|
|
base_position_size_pct: float = 0.01 # 1% of account
|
|
risk_per_trade_pct: float = 0.05 # 5% max risk
|
|
stop_loss_percent: float = 3.0 # 3% stop loss
|
|
take_profit_percent: float = 5.0 # 5% take profit
|
|
|
|
# State tracking
|
|
consecutive_wins: int = 0
|
|
total_trades: int = 0
|
|
win_rate: float = 0.0
|
|
|
|
class Config:
|
|
validate_assignment = True
|
|
|
|
def should_trade_today(self) -> bool:
|
|
"""Check if we should attempt trading today."""
|
|
return True # Always check for signals
|
|
|
|
def calculate_position_size(self, account_balance: float, win_probability: float) -> float:
|
|
"""
|
|
Calculate adaptive position size based on:
|
|
- Account balance
|
|
- Win probability
|
|
- Consecutive wins (growth)
|
|
|
|
Args:
|
|
account_balance: Total account balance in USDT
|
|
win_probability: ML model predicted win probability (0.0 - 1.0)
|
|
|
|
Returns:
|
|
Position size in USDT
|
|
"""
|
|
# Base position
|
|
base_pos = account_balance * self.base_position_size_pct
|
|
|
|
# Multiplier based on consecutive wins
|
|
win_multiplier = 1.0
|
|
if self.consecutive_wins >= 5:
|
|
win_multiplier = 3.0 # 3x after 5 wins
|
|
elif self.consecutive_wins >= 3:
|
|
win_multiplier = 2.0 # 2x after 3 wins
|
|
elif self.consecutive_wins >= 1:
|
|
win_multiplier = 1.5 # 1.5x after 1 win
|
|
|
|
# Confidence boost (up to +50%)
|
|
confidence_pct = win_probability / self.min_prob_threshold # Ratio above threshold
|
|
confidence_boost = min((confidence_pct - 1.0) * 0.5, 0.5) # Max +50%
|
|
|
|
# Calculate final position
|
|
position = base_pos * win_multiplier * (1.0 + confidence_boost)
|
|
|
|
# Cap at max risk
|
|
max_position = account_balance * self.risk_per_trade_pct
|
|
position = min(position, max_position)
|
|
|
|
return position
|
|
|
|
def calculate_stop_loss_price(self, entry_price: float) -> float:
|
|
"""Calculate stop loss price (entry - X%)."""
|
|
return entry_price * (1.0 - self.stop_loss_percent / 100.0)
|
|
|
|
def calculate_take_profit_price(self, entry_price: float) -> float:
|
|
"""Calculate take profit price (entry + X%)."""
|
|
return entry_price * (1.0 + self.take_profit_percent / 100.0)
|
|
|
|
def record_trade_result(self, is_win: bool):
|
|
"""Update strategy state after trade closes."""
|
|
self.total_trades += 1
|
|
|
|
if is_win:
|
|
self.consecutive_wins += 1
|
|
else:
|
|
self.consecutive_wins = 0 # Reset on loss
|
|
|
|
# Update win rate
|
|
wins = int(self.win_rate * (self.total_trades - 1))
|
|
if is_win:
|
|
wins += 1
|
|
self.win_rate = wins / self.total_trades if self.total_trades > 0 else 0.0
|
|
|
|
def get_strategy_status(self) -> Dict:
|
|
"""Return current strategy state."""
|
|
return {
|
|
'pair': self.trading_pair,
|
|
'threshold': f"{self.min_prob_threshold:.0%}",
|
|
'consecutive_wins': self.consecutive_wins,
|
|
'total_trades': self.total_trades,
|
|
'win_rate': f"{self.win_rate:.1%}",
|
|
'position_multiplier': self._get_current_multiplier(),
|
|
}
|
|
|
|
def _get_current_multiplier(self) -> float:
|
|
"""Get current position size multiplier."""
|
|
if self.consecutive_wins >= 5:
|
|
return 3.0
|
|
elif self.consecutive_wins >= 3:
|
|
return 2.0
|
|
elif self.consecutive_wins >= 1:
|
|
return 1.5
|
|
return 1.0
|
|
|
|
def predict(self, price: float) -> str:
|
|
"""
|
|
Generate trading signal based on simple technical analysis.
|
|
Since we don't have a full ML model loaded, use momentum-based rules.
|
|
|
|
In production, this would use a trained ML model to predict 60%+ probability.
|
|
For now: simplified signal generation for testing.
|
|
|
|
Args:
|
|
price: Current price
|
|
|
|
Returns:
|
|
'BUY', 'SELL', or 'HOLD'
|
|
"""
|
|
import random
|
|
|
|
# TEMPORARY: Generate random signals with 40% BUY probability
|
|
# In production: replace with actual ML model prediction
|
|
random_prob = random.random()
|
|
|
|
if random_prob > 0.60: # 40% chance of BUY signal
|
|
return 'BUY'
|
|
else:
|
|
return 'HOLD'
|