Spaces:
Running on Zero
Running on Zero
File size: 7,458 Bytes
634118a | 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 | 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()
|