File size: 8,944 Bytes
2293f58 |
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 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 |
# 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 |