strategies/backtrader.py

# strategies/backtrader.py
import backtrader as bt
import sys
import os
import pandas as pd

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from data.api_client import YahooFinanceClient


class Logger:
    def __init__(self):
        self.logs = []
    
    def add_log(self, log_entry):
        self.logs.append(log_entry)
    
    def get_logs(self):
        return self.logs.copy()  # Return a copy to prevent direct modification

class CustomStrategy(bt.Strategy):
    params = (
        ('analysis', None),
        ('logger', None),        
        ('rsi_period', 14),
        ('rsi_upper', 70),
        ('rsi_lower', 30),
        ('sma_short', 50),
        ('sma_long', 200),
        ('max_loss_percent', 0.02),
        ('take_profit_percent', 0.05),
        ('position_size', 0.1),
        ('atr_period', 14),
        ('atr_multiplier', 3),
        ('sentiment_threshold', 0.6),  # Novo parâmetro
        ('confidence_threshold', 0.7)  # Novo parâmetro
        )

    def __init__(self):
        # Parâmetros agora são acessados via self.params
        self.logger = self.params.logger
        self.recommendation = self.params.analysis['recommendation'] if self.params.analysis else 'HOLD'
        self.technical_analysis = self.params.analysis['technical'] if self.params.analysis else None
        self.sentiment_analysis = self.params.analysis['sentiment'] if self.params.analysis else None
        self.confidence = self.params.analysis['confidence']['total_confidence'] if self.params.analysis else 0.5
        
        # Indicadores usando parâmetros dinâmicos
        self.rsi = bt.indicators.RSI(
            self.data.close, 
            period=self.params.rsi_period
        )
        
        self.sma_short = bt.indicators.SMA(
            self.data.close, 
            period=self.params.sma_short
        )
        
        self.sma_long = bt.indicators.SMA(
            self.data.close, 
            period=self.params.sma_long
        )


        # Technical Indicators
        self.rsi = bt.indicators.RSI(self.data.close, period=self.p.rsi_period)
        self.sma_short = bt.indicators.SMA(self.data.close, period=self.p.sma_short)
        self.sma_long = bt.indicators.SMA(self.data.close, period=self.p.sma_long)
        
        # Volatility Indicator
        self.atr = bt.indicators.ATR(self.data, period=self.p.atr_period)
        
        # Trading management
        self.order = None
        self.stop_price = None
        self.take_profit_price = None
        self.buy_price = None
        self.entry_date = None
            
    def log(self, txt, dt=None):
        dt = dt or self.datas[0].datetime.date(0)
        log_entry = f'{dt.isoformat()}, {txt}'
        self.logger.add_log(log_entry)  # Store in shared logger
        print(log_entry)
        
    def notify_order(self, order):
        if order.status in [order.Submitted, order.Accepted]:
            return

        if order.status in [order.Completed]:
            if order.isbuy():
                self.log(f'BUY EXECUTED, Price: {order.executed.price:.2f}, Cost: {order.executed.value:.2f}, Comm {order.executed.comm:.2f}')
                self.buy_price = order.executed.price
                self.entry_date = self.datas[0].datetime.date(0)
            else:
                self.log(f'SELL EXECUTED, Price: {order.executed.price:.2f}, Cost: {order.executed.value:.2f}, Comm {order.executed.comm:.2f}')
                
            self.order = None

    def notify_trade(self, trade):
        if not trade.isclosed:
            return
        
        self.log(f'TRADE PROFIT, GROSS: {trade.pnl:.2f}, NET: {trade.pnlcomm:.2f}')

    def calculate_position_size(self):
        portfolio_value = self.broker.getvalue()
        return int((portfolio_value * self.p.position_size) / self.data.close[0])

    def next(self):
        # Prevent multiple orders
        if self.order:
            return

        current_price = self.data.close[0]
        
        # Usar parâmetros dinâmicos nas regras
        stop_loss = current_price * (1 - self.params.max_loss_percent)
        take_profit = current_price * (1 + self.params.take_profit_percent)
 
        # Analyze prior analysis for additional confirmation
        analysis_confirmation = self._analyze_prior_research()

        # No open position - look for entry
        if not self.position:
            # Enhanced entry conditions
            # Condições com parâmetros ajustáveis
            entry_conditions = (
                current_price > self.sma_long[0] and
                self.rsi[0] < self.params.rsi_lower and
                bool(self.params.analysis['confidence']['total_confidence'] > self.p.confidence_threshold)
            )

            if entry_conditions:
                # Calculate position size
                size = self.calculate_position_size()
                
                # Place buy order
                self.order = self.buy(size=size)
                
                # Calculate stop loss and take profit
                stop_loss = current_price * (1 - self.p.max_loss_percent)
                take_profit = current_price * (1 + self.p.take_profit_percent)
                
                # Alternative stop loss using ATR for volatility
                atr_stop = current_price - (self.atr[0] * self.p.atr_multiplier)
                self.stop_price = max(stop_loss, atr_stop)
                self.take_profit_price = take_profit

        # Manage existing position
        else:
            # Exit conditions
            exit_conditions = (
                current_price < self.stop_price or  # Stop loss triggered
                current_price > self.take_profit_price or  # Take profit reached
                self.rsi[0] > self.p.rsi_upper or  # Overbought condition
                current_price < self.sma_short[0] or  # Trend change
                not analysis_confirmation  # Loss of analysis confirmation
            )

            if exit_conditions:
                self.close()  # Close entire position
                self.stop_price = None
                self.take_profit_price = None

    def _analyze_prior_research(self):
        # Integrate multiple analysis aspects
        if not self.p.analysis:
            return True

        # Sentiment analysis check
        sentiment_positive = (
            self.sentiment_analysis and 
            self.sentiment_analysis['positive'] > self.p.sentiment_threshold
        )

        # Technical analysis check
        technical_bullish = (
            self.technical_analysis and 
            self.technical_analysis['trend'] == 'bullish'
        )

        # Confidence check
        high_confidence = bool(self.confidence > self.p.confidence_threshold)

        # Combine conditions
        return sentiment_positive and technical_bullish and high_confidence

    def stop(self):
        # Final report when backtest completes
        self.log('Final Portfolio Value: %.2f' % self.broker.getvalue())

class BacktraderIntegration:
    def __init__(self, analysis_result=None, strategy_params=None):
        self.cerebro = bt.Cerebro()
        self.analysis = analysis_result
        self.strategy_params = strategy_params or {}
        self.logger = Logger()  # Create shared logger instance
        self.setup_environment()

    def setup_environment(self):
        self.cerebro.broker.setcash(100000.0)
        self.cerebro.broker.setcommission(commission=0.001)
        self.cerebro.addstrategy(CustomStrategy, analysis=self.analysis, logger=self.logger, **self.strategy_params)

    def add_data_feed(self, ticker, start_date, end_date):
        # Convert datetime to string
        start_str = start_date.strftime("%Y-%m-%d")
        end_str = end_date.strftime("%Y-%m-%d")
        
        # Download data from Yahoo Finance
        df = YahooFinanceClient.download_data(ticker, start_str, end_str, period=None)

                
        if not isinstance(df, pd.DataFrame):
            raise TypeError(f"Esperado pandas.DataFrame, mas recebeu {type(df)}")

        # adjust the columns
        if isinstance(df.columns, pd.MultiIndex):
            df.columns = df.columns.droplevel(1)  # remove the multi-index

        # minimum columns expected
        expected_columns = ["Open", "High", "Low", "Close", "Volume"]

        # Make sure that the columns are correct
        if not all(col in df.columns for col in expected_columns):
            raise ValueError(f"Colunas do DataFrame incorretas: {df.columns}")        
        
        data = bt.feeds.PandasData(dataname=df)
        self.cerebro.adddata(data)

    def run_simulation(self, initial_cash, commission):
        self.cerebro.broker.setcash(initial_cash)
        self.cerebro.broker.setcommission(commission=commission/100)
        self.cerebro.run()
        
        logs = self.logger.get_logs()
        
        return self.cerebro.broker.getvalue(), logs