import pandas as pd import numpy as np import logging logging.basicConfig(level=logging.DEBUG, filename="debug.log", filemode="a") def generate_signals(df, result, volatility_window=14): try: signals_df = pd.DataFrame(index=df.index) signals_df["Price"] = df["value"] signals_df["Signal"] = "Hold" signals_df["Position_Size"] = 0.0 signals_df["Stop_Loss"] = np.nan signals_df["Take_Profit"] = np.nan rsi_key = "rsi_14" macd_key = "macdh_12_26_9" adx_key = "adx_14" pdi_key = "pdi_14" mdi_key = "mdi_14" atr_key = "atr_14" sentiment_key = "sentiment" for i in range(1, len(df)): vote = 0 rsi_signal = macd_signal = adx_signal = sentiment_signal = model_signal = 0 if rsi_key in df.columns and not pd.isna(df[rsi_key].iloc[i]): rsi = df[rsi_key].iloc[i] rsi_signal = 1 if rsi < 50 else -1 if rsi > 50 else 0 vote += rsi_signal logging.debug(f"RSI at {df.index[i]}: value={rsi:.2f}, signal={rsi_signal}") if macd_key in df.columns and not pd.isna(df[macd_key].iloc[i]): macd = df[macd_key].iloc[i] macd_prev = df[macd_key].iloc[i-1] if i > 0 else 0 macd_signal = 1 if macd > 0 and macd_prev <= 0 else -1 if macd < 0 and macd_prev >= 0 else 0 vote += macd_signal logging.debug(f"MACD at {df.index[i]}: value={macd:.2f}, prev={macd_prev:.2f}, signal={macd_signal}") if adx_key in df.columns and pdi_key in df.columns and mdi_key in df.columns: adx = df[adx_key].iloc[i] pdi = df[pdi_key].iloc[i] mdi = df[mdi_key].iloc[i] if not pd.isna(adx) and adx > 20: adx_signal = 1 if pdi > mdi else -1 if mdi > pdi else 0 vote += adx_signal logging.debug(f"ADX at {df.index[i]}: adx={adx:.2f}, pdi={pdi:.2f}, mdi={mdi:.2f}, signal={adx_signal}") if sentiment_key in df.columns and not pd.isna(df[sentiment_key].iloc[i]): sentiment = df[sentiment_key].iloc[i] sentiment_signal = 1 if sentiment > 0.1 else -1 if sentiment < -0.1 else 0 vote += sentiment_signal logging.debug(f"Sentiment at {df.index[i]}: value={sentiment:.2f}, signal={sentiment_signal}") if "forecast" in result and len(result["forecast"]) > i: forecast = result["forecast"][i] actual = df["value"].iloc[i] model_signal = 1 if forecast > actual * 1.01 else -1 if forecast < actual * 0.99 else 0 vote += model_signal logging.debug(f"Model at {df.index[i]}: forecast={forecast:.2f}, actual={actual:.2f}, signal={model_signal}") signals_df.loc[df.index[i], "Signal"] = "Buy" if vote >= 2 else "Sell" if vote <= -2 else "Hold" signals_df.loc[df.index[i], "Position_Size"] = min(0.1 * abs(vote), 1.0) if atr_key in df.columns and not pd.isna(df[atr_key].iloc[i]): atr = df[atr_key].iloc[i] signals_df.loc[df.index[i], "Stop_Loss"] = df["value"].iloc[i] - 2 * atr if vote >= 2 else df["value"].iloc[i] + 2 * atr if vote <= -2 else np.nan signals_df.loc[df.index[i], "Take_Profit"] = df["value"].iloc[i] + 3 * atr if vote >= 2 else df["value"].iloc[i] - 3 * atr if vote <= -2 else np.nan current_signal = signals_df.iloc[i]["Signal"] logging.debug(f"Signal at {df.index[i]}: RSI={rsi_signal}, MACD={macd_signal}, ADX={adx_signal}, Sentiment={sentiment_signal}, Model={model_signal}, Vote={vote}, Signal={current_signal}") trades_df, equity_df = backtest_signals(signals_df, df) signals_df["Equity"] = equity_df["Equity"] signal_counts = signals_df["Signal"].value_counts().to_dict() total = sum(signal_counts.values()) signal_dist = {k: f"{v} ({v/total*100:.2f}%)" for k, v in signal_counts.items()} signal_dist_str = ", ".join([f'{k}={v}' for k, v in signal_dist.items()]) logging.info(f"Signal distribution: {signal_dist_str}") logging.info(f"Signals generated: {signal_counts}") return signals_df, trades_df, equity_df except Exception as e: logging.error(f"Error in generate_signals: {e}") return pd.DataFrame(), pd.DataFrame(), pd.DataFrame() def backtest_signals(signals_df, df, initial_balance=10000): try: balance = initial_balance position = 0 trades = [] equity_curve = [balance] entry_price = 0 # Iterate through the original DataFrame's index to ensure equity_curve aligns for idx, row in df.iterrows(): # Find the corresponding signal for this date signal_row = signals_df.loc[idx] if idx in signals_df.index else None if signal_row is not None: price = signal_row["Price"] signal = signal_row["Signal"] position_size = signal_row["Position_Size"] stop_loss = signal_row["Stop_Loss"] take_profit = signal_row["Take_Profit"] if signal == "Buy" and position == 0: shares = position_size * balance / price position = shares entry_price = price trades.append({"Date": str(idx.date()), "Type": "Buy", "Price": price, "Shares": shares}) logging.debug(f"Buy at {price:.2f}, Shares: {shares:.2f}") elif signal == "Sell" and position > 0: balance += position * (price - entry_price) trades.append({"Date": str(idx.date()), "Type": "Sell", "Price": price, "Shares": position, "Profit": position * (price - entry_price)}) position = 0 profit_val = trades[-1]["Profit"] logging.debug(f"Sell at {price:.2f}, Profit: {profit_val:.2f}") if position > 0 and not pd.isna(stop_loss) and not pd.isna(take_profit): if price <= stop_loss or price >= take_profit: balance += position * (price - entry_price) trades.append({"Date": str(idx.date()), "Type": "Exit", "Price": price, "Shares": position, "Profit": position * (price - entry_price)}) position = 0 profit_val = trades[-1]["Profit"] logging.debug(f"Exit at {price:.2f}, Profit: {profit_val:.2f}") current_equity = balance + position * (row["value"] - entry_price) if position > 0 else balance equity_curve.append(current_equity) # The first element of equity_curve is the initial balance, remove it to align with df.index equity_curve = equity_curve[1:] trades_df = pd.DataFrame(trades) equity_df = pd.DataFrame({"Equity": equity_curve}, index=df.index) logging.info(f"Backtest completed: {len(trades)} trades, Final Balance: {balance:.2f}") return trades_df, equity_df except Exception as e: logging.error(f"Backtest error: {e}") return pd.DataFrame(), pd.DataFrame(), pd.DataFrame(), pd.DataFrame()