114 lines
3.8 KiB
Python
114 lines
3.8 KiB
Python
|
import pandas as pd
|
||
|
|
import ta
|
||
|
|
import logging
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
class TechnicalAgent:
|
||
|
|
MIN_CANDLES = 30
|
||
|
|
|
||
|
|
def analyze(self, df: pd.DataFrame) -> float:
|
||
|
|
if len(df) < self.MIN_CANDLES:
|
||
|
|
return 50.0
|
||
|
|
|
||
|
|
try:
|
||
|
|
signals = []
|
||
|
|
|
||
|
|
# RSI (14)
|
||
|
|
rsi = ta.momentum.RSIIndicator(df["close"], window=14).rsi().iloc[-1]
|
||
|
|
if rsi < 30:
|
||
|
|
signals.append(65)
|
||
|
|
elif rsi < 45:
|
||
|
|
signals.append(70)
|
||
|
|
elif rsi < 55:
|
||
|
|
signals.append(50)
|
||
|
|
elif rsi < 70:
|
||
|
|
signals.append(35)
|
||
|
|
else:
|
||
|
|
signals.append(40)
|
||
|
|
|
||
|
|
# MACD
|
||
|
|
macd_ind = ta.trend.MACD(df["close"])
|
||
|
|
macd_diff = macd_ind.macd_diff().iloc[-1]
|
||
|
|
prev_diff = macd_ind.macd_diff().iloc[-2]
|
||
|
|
if macd_diff > 0 and prev_diff <= 0:
|
||
|
|
signals.append(85)
|
||
|
|
elif macd_diff < 0 and prev_diff >= 0:
|
||
|
|
signals.append(15)
|
||
|
|
elif macd_diff > 0:
|
||
|
|
signals.append(65)
|
||
|
|
else:
|
||
|
|
signals.append(35)
|
||
|
|
|
||
|
|
# Bollinger Bands
|
||
|
|
bb = ta.volatility.BollingerBands(df["close"])
|
||
|
|
bb_high = bb.bollinger_hband().iloc[-1]
|
||
|
|
bb_low = bb.bollinger_lband().iloc[-1]
|
||
|
|
price = df["close"].iloc[-1]
|
||
|
|
bb_pct = (price - bb_low) / (bb_high - bb_low) if (bb_high - bb_low) > 0 else 0.5
|
||
|
|
if bb_pct < 0.2:
|
||
|
|
signals.append(55)
|
||
|
|
elif bb_pct > 0.8:
|
||
|
|
signals.append(40)
|
||
|
|
else:
|
||
|
|
signals.append(50)
|
||
|
|
|
||
|
|
# SMA trend (20 vs 50)
|
||
|
|
sma20 = df["close"].rolling(20).mean().iloc[-1]
|
||
|
|
sma50 = df["close"].rolling(50).mean().iloc[-1] if len(df) >= 50 else sma20
|
||
|
|
if price > sma20 > sma50:
|
||
|
|
signals.append(80)
|
||
|
|
elif price > sma20:
|
||
|
|
signals.append(65)
|
||
|
|
elif price < sma20 < sma50:
|
||
|
|
signals.append(20)
|
||
|
|
else:
|
||
|
|
signals.append(40)
|
||
|
|
|
||
|
|
# SMA 200 (long-term trend)
|
||
|
|
if len(df) >= 200:
|
||
|
|
sma200 = df["close"].rolling(200).mean().iloc[-1]
|
||
|
|
if price > sma200:
|
||
|
|
signals.append(70)
|
||
|
|
else:
|
||
|
|
signals.append(30)
|
||
|
|
|
||
|
|
# Volume trend
|
||
|
|
vol_avg = df["volume"].rolling(20).mean().iloc[-1]
|
||
|
|
vol_current = df["volume"].iloc[-1]
|
||
|
|
vol_ratio = vol_current / vol_avg if vol_avg > 0 else 1.0
|
||
|
|
if vol_ratio > 2.0 and price > sma20:
|
||
|
|
signals.append(80)
|
||
|
|
elif vol_ratio > 2.0:
|
||
|
|
signals.append(40)
|
||
|
|
else:
|
||
|
|
signals.append(50)
|
||
|
|
|
||
|
|
# OBV trend
|
||
|
|
obv = ta.volume.OnBalanceVolumeIndicator(df["close"], df["volume"]).on_balance_volume()
|
||
|
|
obv_sma = obv.rolling(20).mean()
|
||
|
|
if obv.iloc[-1] > obv_sma.iloc[-1]:
|
||
|
|
signals.append(65)
|
||
|
|
else:
|
||
|
|
signals.append(35)
|
||
|
|
|
||
|
|
# Candlestick patterns (simplified)
|
||
|
|
last = df.iloc[-1]
|
||
|
|
prev = df.iloc[-2]
|
||
|
|
body = last["close"] - last["open"]
|
||
|
|
prev_body = prev["close"] - prev["open"]
|
||
|
|
if prev_body < 0 and body > 0 and body > abs(prev_body):
|
||
|
|
signals.append(80) # bullish engulfing
|
||
|
|
elif prev_body > 0 and body < 0 and abs(body) > prev_body:
|
||
|
|
signals.append(20) # bearish engulfing
|
||
|
|
elif abs(body) < (last["high"] - last["low"]) * 0.1:
|
||
|
|
signals.append(50) # doji
|
||
|
|
else:
|
||
|
|
signals.append(50)
|
||
|
|
|
||
|
|
return round(sum(signals) / len(signals), 1)
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
logger.error(f"Technical analysis error: {e}")
|
||
|
|
return 50.0
|