Spaces:
Running
Running
| import numpy as np | |
| class TradingLogic: | |
| def __init__(self): | |
| self.risk_reward_ratio = 2.0 # 1:2 risk/reward | |
| def generate_signal(self, predictions, current_price, df): | |
| """Generate buy/sell signal based on predictions and indicators""" | |
| try: | |
| if len(predictions) < 5: | |
| return "hold", 0.0 | |
| # Get last values | |
| rsi = df['RSI'].iloc[-1] if 'RSI' in df.columns else 50 | |
| macd = df['MACD'].iloc[-1] if 'MACD' in df.columns else 0 | |
| macd_signal = df['MACD_signal'].iloc[-1] if 'MACD_signal' in df.columns else 0 | |
| # Prediction trend | |
| pred_trend = np.polyfit(range(len(predictions)), predictions, 1)[0] | |
| # Price vs predictions | |
| pred_mean = np.mean(predictions) | |
| price_diff_pct = (pred_mean - current_price) / current_price if current_price != 0 else 0 | |
| # Initialize signals | |
| signals = [] | |
| confidences = [] | |
| # RSI signal | |
| if rsi < 30: | |
| signals.append("buy") | |
| confidences.append(0.6) | |
| elif rsi > 70: | |
| signals.append("sell") | |
| confidences.append(0.6) | |
| # MACD signal | |
| if macd > macd_signal and macd < 0: | |
| signals.append("buy") | |
| confidences.append(0.7) | |
| elif macd < macd_signal and macd > 0: | |
| signals.append("sell") | |
| confidences.append(0.7) | |
| # Prediction signal | |
| if pred_trend > 0 and price_diff_pct > 0.01: | |
| signals.append("buy") | |
| confidences.append(min(abs(price_diff_pct) * 10, 0.8)) | |
| elif pred_trend < 0 and price_diff_pct < -0.01: | |
| signals.append("sell") | |
| confidences.append(min(abs(price_diff_pct) * 10, 0.8)) | |
| # Bollinger Bands | |
| if 'BB_lower' in df.columns and 'BB_upper' in df.columns: | |
| bb_position = (current_price - df['BB_lower'].iloc[-1]) / (df['BB_upper'].iloc[-1] - df['BB_lower'].iloc[-1]) | |
| if bb_position < 0.2: | |
| signals.append("buy") | |
| confidences.append(0.5) | |
| elif bb_position > 0.8: | |
| signals.append("sell") | |
| confidences.append(0.5) | |
| # Aggregate signals | |
| if not signals: | |
| return "hold", 0.0 | |
| buy_count = signals.count("buy") | |
| sell_count = signals.count("sell") | |
| total_signals = len(signals) | |
| if buy_count > sell_count and buy_count >= total_signals * 0.5: | |
| signal = "buy" | |
| confidence = np.mean([c for s, c in zip(signals, confidences) if s == "buy"]) | |
| elif sell_count > buy_count and sell_count >= total_signals * 0.5: | |
| signal = "sell" | |
| confidence = np.mean([c for s, c in zip(signals, confidences) if s == "sell"]) | |
| else: | |
| signal = "hold" | |
| confidence = 0.0 | |
| return signal, confidence | |
| except Exception as e: | |
| print(f"Signal generation error: {e}") | |
| return "hold", 0.0 | |
| def calculate_tp_sl(self, current_price, atr, signal): | |
| """Calculate Take Profit and Stop Loss based on ATR""" | |
| try: | |
| if signal == "hold": | |
| return None, None | |
| # Use 2x ATR for stop loss, 4x ATR for take profit (1:2 ratio) | |
| sl_distance = atr * 2 if atr else current_price * 0.02 | |
| tp_distance = atr * 4 if atr else current_price * 0.04 | |
| if signal == "buy": | |
| tp = current_price + tp_distance | |
| sl = current_price - sl_distance | |
| else: # sell | |
| tp = current_price - tp_distance | |
| sl = current_price + sl_distance | |
| return round(tp, 2), round(sl, 2) | |
| except Exception as e: | |
| print(f"TP/SL calculation error: {e}") | |
| return None, None |