65 lines
2.0 KiB
Python
Executable File
65 lines
2.0 KiB
Python
Executable File
from datetime import datetime, timedelta
|
|
from typing import Optional
|
|
from pydantic import BaseModel
|
|
|
|
class DCAStrategy(BaseModel):
|
|
"""Dollar-Cost-Averaging strategy configuration and logic."""
|
|
|
|
trading_pair: str # e.g., "BTCUSDT"
|
|
dca_amount_usd: float # Amount to invest per cycle
|
|
interval_hours: float # Time between buys
|
|
stop_loss_percent: float # Stop loss percentage
|
|
|
|
class Config:
|
|
validate_assignment = True
|
|
|
|
def should_execute_dca(self, last_order_time: Optional[datetime] = None) -> bool:
|
|
"""
|
|
Determine if DCA order should execute.
|
|
|
|
Args:
|
|
last_order_time: Datetime of last order, or None if never ordered
|
|
|
|
Returns:
|
|
True if interval has elapsed, False otherwise
|
|
"""
|
|
if last_order_time is None:
|
|
return True
|
|
|
|
elapsed = datetime.utcnow() - last_order_time
|
|
interval = timedelta(hours=self.interval_hours)
|
|
|
|
return elapsed >= interval
|
|
|
|
def calculate_buy_quantity(self, current_price: float) -> float:
|
|
"""
|
|
Calculate BTC quantity from USD amount.
|
|
|
|
Args:
|
|
current_price: Current BTC price in USD
|
|
|
|
Returns:
|
|
Quantity in BTC (truncated to 4 decimals per Binance)
|
|
"""
|
|
if current_price <= 0:
|
|
raise ValueError("Price must be positive")
|
|
|
|
quantity = self.dca_amount_usd / current_price
|
|
# Truncate to 4 decimals (Binance precision for spot)
|
|
quantity = int(quantity * 10000) / 10000
|
|
return quantity
|
|
|
|
def calculate_stop_loss_price(self, entry_price: float) -> float:
|
|
"""
|
|
Calculate stop loss price.
|
|
|
|
Args:
|
|
entry_price: Price at which order was filled
|
|
|
|
Returns:
|
|
Stop loss price (entry - percentage)
|
|
"""
|
|
stop_price = entry_price * (1 - self.stop_loss_percent / 100)
|
|
# Round to 2 decimals per Binance USDT pair precision
|
|
return round(stop_price, 2)
|