File size: 4,245 Bytes
590a501 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | import asyncio
import logging
from app.models.schemas import StrategyConfig, StrategyStatus
from app.services.market_data import market_data_service
from app.services.order_manager import order_manager
from app.strategies.base import BaseStrategy
from app.strategies.bollinger_bands import BollingerBandsStrategy
from app.strategies.dual_thrust import DualThrustStrategy
from app.strategies.ma_crossover import MACrossoverStrategy
logger = logging.getLogger(__name__)
STRATEGY_CLASSES: dict[str, type[BaseStrategy]] = {
"ma_crossover": MACrossoverStrategy,
"bollinger_bands": BollingerBandsStrategy,
"dual_thrust": DualThrustStrategy,
}
class StrategyEngine:
def __init__(self):
self._strategies: dict[str, BaseStrategy] = {}
self._configs: dict[str, StrategyConfig] = {}
self._tasks: dict[str, asyncio.Task] = {}
def get_available_strategies(self) -> list[dict]:
result = []
for key, cls in STRATEGY_CLASSES.items():
instance = cls(strategy_id="temp", symbol="")
result.append({
"type": key,
"name": instance.name,
"description": instance.description,
"default_params": instance.default_params,
})
return result
def get_running_strategies(self) -> list[StrategyConfig]:
return list(self._configs.values())
def get_strategy_signals(self, strategy_id: str) -> list[dict]:
strategy = self._strategies.get(strategy_id)
if strategy:
return strategy.signals
return []
def get_strategy_performance(self, strategy_id: str):
strategy = self._strategies.get(strategy_id)
if strategy:
return strategy.performance
return None
def add_strategy(self, config: StrategyConfig) -> StrategyConfig:
cls = STRATEGY_CLASSES.get(config.strategy_type)
if cls is None:
raise ValueError(f"Unknown strategy type: {config.strategy_type}")
strategy = cls(
strategy_id=config.strategy_id,
symbol=config.symbol,
params=config.params,
)
self._strategies[config.strategy_id] = strategy
self._configs[config.strategy_id] = config
return config
def start_strategy(self, strategy_id: str) -> StrategyConfig | None:
config = self._configs.get(strategy_id)
if config is None:
return None
config.status = StrategyStatus.RUNNING
task = asyncio.create_task(self._run_strategy(strategy_id))
self._tasks[strategy_id] = task
return config
def stop_strategy(self, strategy_id: str) -> StrategyConfig | None:
config = self._configs.get(strategy_id)
if config is None:
return None
config.status = StrategyStatus.STOPPED
task = self._tasks.pop(strategy_id, None)
if task:
task.cancel()
return config
def remove_strategy(self, strategy_id: str) -> bool:
self.stop_strategy(strategy_id)
self._strategies.pop(strategy_id, None)
self._configs.pop(strategy_id, None)
return True
async def _run_strategy(self, strategy_id: str):
strategy = self._strategies.get(strategy_id)
config = self._configs.get(strategy_id)
if not strategy or not config:
return
try:
while config.status == StrategyStatus.RUNNING:
klines = market_data_service.get_kline_history(strategy.symbol)
if klines:
signal = strategy.calculate_signal(klines)
if signal:
order = order_manager.place_order(signal)
logger.info(
"Strategy %s generated %s signal for %s, order: %s",
strategy_id, signal.side, signal.symbol, order.order_id
)
await asyncio.sleep(2)
except asyncio.CancelledError:
pass
except Exception as e:
logger.error("Strategy %s error: %s", strategy_id, e)
config.status = StrategyStatus.ERROR
strategy_engine = StrategyEngine()
|