core/signals.py

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()