app.backbone.services.strategy_service 

 1from typing import List
 2from sqlalchemy import UUID
 3from app.backbone.entities.bot import Bot
 4from app.backbone.database.db_service import DbService
 5from app.backbone.entities.strategy import Strategy
 6from app.backbone.services.backtest_service import BacktestService
 7from app.backbone.services.bot_service import BotService
 8from app.backbone.services.operation_result import OperationResult
 9from app.backbone.utils.general_purpose import profile_function
10
11class StrategyService:
12    def __init__(self):
13        self.db_service = DbService()
14        self.backtest_service = BacktestService()
15        self.bot_service = BotService()
16
17        
18    def create(self, name:str, description:str, metatrader_name:str) -> OperationResult:
19        with self.db_service.get_database() as db:
20            
21            strategy_by_filter = self.db_service.get_by_filter(db, Strategy, Name=name)
22            
23            if strategy_by_filter is None:
24                new_strategy = Strategy(Name=name, Description=description, MetaTraderName=metatrader_name)
25                strategy = self.db_service.create(db, new_strategy)
26                result = OperationResult(ok=True, message=None, item=strategy)
27                
28                return result
29            
30            result = OperationResult(ok=False, message='El item ya esta cargado en la BD', item=None)
31            return result
32            
33    def get_all(self) -> List[Strategy]:
34        with self.db_service.get_database() as db:
35            all_strategies = self.db_service.get_all(db, Strategy)
36            return all_strategies
37
38    def delete(self, strategy_id) -> OperationResult:
39        bots = self.bot_service.get_bots_by_strategy(strategy_id=strategy_id)
40        for bot in bots:
41            for backtest in bot.BotPerformance:
42                _ = self.backtest_service.delete(backtest.Id)
43
44        with self.db_service.get_database() as db:
45        
46            strategy = self.db_service.delete(db, Strategy, strategy_id)
47            result = OperationResult(ok=True, message=None, item=strategy)
48            return result
49    
50    def get_by_id(self, id) -> Strategy:
51        with self.db_service.get_database() as db:
52            strategy = self.db_service.get_by_id(db, Strategy, id)
53            return strategy
54    
55    def update(self, id:int, name:str, description:str, metatrader_name:str) -> OperationResult:
56        with self.db_service.get_database() as db:
57            new_strategy = Strategy(Id=id, Name=name, Description=description, MetaTraderName=metatrader_name)
58            
59            strategy = self.db_service.update(db, Strategy, new_strategy)
60            
61            result = OperationResult(ok=True, message=None, item=strategy)
62            return result
63
64    def get_used_strategies(self) -> List[Strategy]:
65        with self.db_service.get_database() as db:
66            strategies = (
67                db.query(Strategy)
68                .join(Bot, Strategy.Id == Bot.StrategyId)  # Relación entre Strategy y Bot
69                .distinct()  # Evita duplicados
70                .all()  # Recupera los resultados
71            )
72            
73            return strategies
class StrategyService: 
12class StrategyService:
13    def __init__(self):
14        self.db_service = DbService()
15        self.backtest_service = BacktestService()
16        self.bot_service = BotService()
17
18        
19    def create(self, name:str, description:str, metatrader_name:str) -> OperationResult:
20        with self.db_service.get_database() as db:
21            
22            strategy_by_filter = self.db_service.get_by_filter(db, Strategy, Name=name)
23            
24            if strategy_by_filter is None:
25                new_strategy = Strategy(Name=name, Description=description, MetaTraderName=metatrader_name)
26                strategy = self.db_service.create(db, new_strategy)
27                result = OperationResult(ok=True, message=None, item=strategy)
28                
29                return result
30            
31            result = OperationResult(ok=False, message='El item ya esta cargado en la BD', item=None)
32            return result
33            
34    def get_all(self) -> List[Strategy]:
35        with self.db_service.get_database() as db:
36            all_strategies = self.db_service.get_all(db, Strategy)
37            return all_strategies
38
39    def delete(self, strategy_id) -> OperationResult:
40        bots = self.bot_service.get_bots_by_strategy(strategy_id=strategy_id)
41        for bot in bots:
42            for backtest in bot.BotPerformance:
43                _ = self.backtest_service.delete(backtest.Id)
44
45        with self.db_service.get_database() as db:
46        
47            strategy = self.db_service.delete(db, Strategy, strategy_id)
48            result = OperationResult(ok=True, message=None, item=strategy)
49            return result
50    
51    def get_by_id(self, id) -> Strategy:
52        with self.db_service.get_database() as db:
53            strategy = self.db_service.get_by_id(db, Strategy, id)
54            return strategy
55    
56    def update(self, id:int, name:str, description:str, metatrader_name:str) -> OperationResult:
57        with self.db_service.get_database() as db:
58            new_strategy = Strategy(Id=id, Name=name, Description=description, MetaTraderName=metatrader_name)
59            
60            strategy = self.db_service.update(db, Strategy, new_strategy)
61            
62            result = OperationResult(ok=True, message=None, item=strategy)
63            return result
64
65    def get_used_strategies(self) -> List[Strategy]:
66        with self.db_service.get_database() as db:
67            strategies = (
68                db.query(Strategy)
69                .join(Bot, Strategy.Id == Bot.StrategyId)  # Relación entre Strategy y Bot
70                .distinct()  # Evita duplicados
71                .all()  # Recupera los resultados
72            )
73            
74            return strategies
db_service
backtest_service
bot_service
def create( self, name: str, description: str, metatrader_name: str) -> app.backbone.services.operation_result.OperationResult: 
19    def create(self, name:str, description:str, metatrader_name:str) -> OperationResult:
20        with self.db_service.get_database() as db:
21            
22            strategy_by_filter = self.db_service.get_by_filter(db, Strategy, Name=name)
23            
24            if strategy_by_filter is None:
25                new_strategy = Strategy(Name=name, Description=description, MetaTraderName=metatrader_name)
26                strategy = self.db_service.create(db, new_strategy)
27                result = OperationResult(ok=True, message=None, item=strategy)
28                
29                return result
30            
31            result = OperationResult(ok=False, message='El item ya esta cargado en la BD', item=None)
32            return result
def get_all(self) -> List[app.backbone.entities.strategy.Strategy]: 
34    def get_all(self) -> List[Strategy]:
35        with self.db_service.get_database() as db:
36            all_strategies = self.db_service.get_all(db, Strategy)
37            return all_strategies
def delete( self, strategy_id) -> app.backbone.services.operation_result.OperationResult: 
39    def delete(self, strategy_id) -> OperationResult:
40        bots = self.bot_service.get_bots_by_strategy(strategy_id=strategy_id)
41        for bot in bots:
42            for backtest in bot.BotPerformance:
43                _ = self.backtest_service.delete(backtest.Id)
44
45        with self.db_service.get_database() as db:
46        
47            strategy = self.db_service.delete(db, Strategy, strategy_id)
48            result = OperationResult(ok=True, message=None, item=strategy)
49            return result
def get_by_id(self, id) -> app.backbone.entities.strategy.Strategy: 
51    def get_by_id(self, id) -> Strategy:
52        with self.db_service.get_database() as db:
53            strategy = self.db_service.get_by_id(db, Strategy, id)
54            return strategy
def update( self, id: int, name: str, description: str, metatrader_name: str) -> app.backbone.services.operation_result.OperationResult: 
56    def update(self, id:int, name:str, description:str, metatrader_name:str) -> OperationResult:
57        with self.db_service.get_database() as db:
58            new_strategy = Strategy(Id=id, Name=name, Description=description, MetaTraderName=metatrader_name)
59            
60            strategy = self.db_service.update(db, Strategy, new_strategy)
61            
62            result = OperationResult(ok=True, message=None, item=strategy)
63            return result
def get_used_strategies(self) -> List[app.backbone.entities.strategy.Strategy]: 
65    def get_used_strategies(self) -> List[Strategy]:
66        with self.db_service.get_database() as db:
67            strategies = (
68                db.query(Strategy)
69                .join(Bot, Strategy.Id == Bot.StrategyId)  # Relación entre Strategy y Bot
70                .distinct()  # Evita duplicados
71                .all()  # Recupera los resultados
72            )
73            
74            return strategies