206 lines
8.6 KiB
Python
206 lines
8.6 KiB
Python
"""
|
|
Module for Technical Analysis using FinanceToolkit.
|
|
Calculates RSI, EMA, Volatility, MACD, Bollinger Bands, and Fibonacci Retracement Levels.
|
|
"""
|
|
from financetoolkit import Toolkit
|
|
import pandas as pd
|
|
import numpy as np
|
|
|
|
def get_technicals(ticker, api_key, period="daily"):
|
|
"""
|
|
Fetches comprehensive technical indicators for a given ticker.
|
|
|
|
Args:
|
|
ticker (str): Stock ticker symbol.
|
|
api_key (str): FMP API Key.
|
|
period (str): 'daily', 'weekly', etc.
|
|
|
|
Returns:
|
|
dict: Latest technical values and signals for 'Pro' analysis.
|
|
"""
|
|
try:
|
|
companies = Toolkit(
|
|
tickers=[ticker],
|
|
api_key=api_key,
|
|
)
|
|
|
|
# Helper to safely get the latest value for the specific ticker
|
|
def get_latest_series(df, col_name=None):
|
|
if df.empty: return 0
|
|
# multi-index handling: typically (Symbol, Metric) or (Metric, Symbol)
|
|
if isinstance(df.columns, pd.MultiIndex):
|
|
# Try selecting ticker from level 1
|
|
if ticker in df.columns.get_level_values(1):
|
|
df = df.xs(ticker, level=1, axis=1)
|
|
# Try selecting ticker from level 0
|
|
elif ticker in df.columns.get_level_values(0):
|
|
df = df.xs(ticker, level=0, axis=1)
|
|
|
|
# If still DataFrame, try to pick the specific column or just the first
|
|
if isinstance(df, pd.DataFrame):
|
|
if col_name and col_name in df.columns:
|
|
return df[col_name].iloc[-1]
|
|
return df.iloc[-1, 0] # Default to first column
|
|
else:
|
|
return df.iloc[-1]
|
|
|
|
def get_value(val):
|
|
return val.values[0] if hasattr(val, 'values') else val
|
|
|
|
# 1. RSI (14 periods) - Good for Overbought/Oversold
|
|
rsi_df = companies.technicals.get_relative_strength_index(window=14)
|
|
current_rsi = get_value(get_latest_series(rsi_df))
|
|
|
|
# 2. EMA (50 periods) - Trend Filter
|
|
ema_df = companies.technicals.get_exponential_moving_average(window=50)
|
|
current_ema = get_value(get_latest_series(ema_df))
|
|
|
|
# 2b. EMA (10 periods) - Short Term Trend (Phase 3)
|
|
ema10_df = companies.technicals.get_exponential_moving_average(window=10)
|
|
current_ema10 = get_value(get_latest_series(ema10_df))
|
|
|
|
# 2c. EMA (5 periods) - Ultra-Short Term (Phase 5 Aggressive)
|
|
ema5_df = companies.technicals.get_exponential_moving_average(window=5)
|
|
current_ema5 = get_value(get_latest_series(ema5_df))
|
|
|
|
# 3. MACD (12, 26, 9) - Momentum
|
|
macd_df = companies.technicals.get_moving_average_convergence_divergence(short_window=12, long_window=26, signal_window=9)
|
|
# MACD usually returns MACD line and Signal line. We need both.
|
|
# Structure likely: [MACD, Signal, Histogram] columns per ticker
|
|
# Let's get the full series for the ticker first
|
|
if isinstance(macd_df.columns, pd.MultiIndex):
|
|
if ticker in macd_df.columns.get_level_values(1):
|
|
macd_data = macd_df.xs(ticker, level=1, axis=1)
|
|
elif ticker in macd_df.columns.get_level_values(0):
|
|
macd_data = macd_df.xs(ticker, level=0, axis=1)
|
|
else:
|
|
macd_data = macd_df
|
|
else:
|
|
macd_data = macd_df
|
|
|
|
# Assuming columns like 'MACD', 'Signal', 'Histogram' or similar exist
|
|
# If columns are just numbered or named differently, we take last row
|
|
if 'MACD' in macd_data.columns and 'Signal' in macd_data.columns:
|
|
current_macd = macd_data['MACD'].iloc[-1]
|
|
current_signal_line = macd_data['Signal'].iloc[-1]
|
|
else:
|
|
# Fallback: assume col 0 is MACD, col 1 is Signal (standard toolkit output order often)
|
|
current_macd = macd_data.iloc[-1, 0]
|
|
current_signal_line = macd_data.iloc[-1, 1] if macd_data.shape[1] > 1 else 0
|
|
|
|
# 4. Bollinger Bands (20, 2) - Volatility & Mean Reversion
|
|
bb_df = companies.technicals.get_bollinger_bands(window=20, num_std_dev=2)
|
|
# Usually returns: Upper, Lower, (maybe Middle)
|
|
if isinstance(bb_df.columns, pd.MultiIndex):
|
|
if ticker in bb_df.columns.get_level_values(1):
|
|
bb_data = bb_df.xs(ticker, level=1, axis=1)
|
|
elif ticker in bb_df.columns.get_level_values(0):
|
|
bb_data = bb_df.xs(ticker, level=0, axis=1)
|
|
else:
|
|
bb_data = bb_df
|
|
else:
|
|
bb_data = bb_df
|
|
|
|
# Inspect columns for Upper/Lower
|
|
# Typ. 'Bollinger High', 'Bollinger Low' or similar
|
|
# Fallback to column indices if names vary
|
|
if bb_data.shape[1] >= 2:
|
|
bb_upper = bb_data.iloc[-1, 0] # Often Upper is first or checking names
|
|
bb_lower = bb_data.iloc[-1, 1]
|
|
# Try to be more specific if possible, but for now robust fallback
|
|
# Let's look for 'Upper' or 'High' in columns
|
|
for col in bb_data.columns:
|
|
if 'Upper' in str(col) or 'High' in str(col): bb_upper = bb_data[col].iloc[-1]
|
|
if 'Lower' in str(col) or 'Low' in str(col): bb_lower = bb_data[col].iloc[-1]
|
|
else:
|
|
bb_upper = 0
|
|
bb_lower = 0
|
|
|
|
# 5. Volatility & Fibonacci & NEW EXPERT METRICS (SMA, Volume, ATR)
|
|
hist_data = companies.get_historical_data()
|
|
|
|
atr = 0
|
|
rvol = 0
|
|
sma_200 = 0
|
|
sma_50 = 0
|
|
|
|
if not hist_data.empty:
|
|
# Handle MultiIndex
|
|
if isinstance(hist_data.columns, pd.MultiIndex):
|
|
try:
|
|
if ticker in hist_data.columns.get_level_values(1):
|
|
hist_data = hist_data.xs(ticker, level=1, axis=1)
|
|
elif ticker in hist_data.columns.get_level_values(0):
|
|
hist_data = hist_data.xs(ticker, level=0, axis=1)
|
|
except: pass
|
|
|
|
daily_returns = hist_data['Return'].iloc[-20:]
|
|
volatility = daily_returns.std() * np.sqrt(252) * 100
|
|
current_close = hist_data['Close'].iloc[-1]
|
|
if isinstance(current_close, pd.Series): current_close = current_close.iloc[0]
|
|
|
|
# --- EXPERT METRICS ---
|
|
# SMA 50 & 200
|
|
sma_50 = hist_data['Close'].rolling(window=50).mean().iloc[-1]
|
|
sma_200 = hist_data['Close'].rolling(window=200).mean().iloc[-1]
|
|
|
|
# Volume Relative Strength (RVOL)
|
|
# Current Volume / Avg Volume 20
|
|
curr_vol = hist_data['Volume'].iloc[-1]
|
|
avg_vol_20 = hist_data['Volume'].rolling(window=20).mean().iloc[-1]
|
|
rvol = curr_vol / avg_vol_20 if avg_vol_20 > 0 else 0
|
|
|
|
# ATR (14)
|
|
high = hist_data['High']
|
|
low = hist_data['Low']
|
|
close = hist_data['Close']
|
|
tr1 = high - low
|
|
tr2 = (high - close.shift()).abs()
|
|
tr3 = (low - close.shift()).abs()
|
|
tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
|
|
atr = tr.rolling(window=14).mean().iloc[-1]
|
|
|
|
# Fibonacci
|
|
last_year = hist_data.iloc[-252:]
|
|
high_price = last_year['High'].max()
|
|
low_price = last_year['Low'].min()
|
|
if isinstance(high_price, pd.Series): high_price = high_price.iloc[0]
|
|
if isinstance(low_price, pd.Series): low_price = low_price.iloc[0]
|
|
|
|
diff = high_price - low_price
|
|
fib_levels = {
|
|
'0.0%': high_price,
|
|
'23.6%': high_price - 0.236 * diff,
|
|
'38.2%': high_price - 0.382 * diff,
|
|
'50.0%': high_price - 0.5 * diff,
|
|
'61.8%': high_price - 0.618 * diff,
|
|
'100.0%': low_price
|
|
}
|
|
else:
|
|
volatility = 0
|
|
current_close = 0
|
|
fib_levels = {}
|
|
|
|
return {
|
|
'symbol': ticker,
|
|
'price': current_close,
|
|
'rsi': current_rsi,
|
|
'ema_50': current_ema,
|
|
'ema_10': current_ema10,
|
|
'ema_5': current_ema5,
|
|
'sma_50': sma_50,
|
|
'sma_200': sma_200,
|
|
'rvol': rvol,
|
|
'atr': atr,
|
|
'macd_line': current_macd,
|
|
'macd_signal': current_signal_line,
|
|
'bb_upper': bb_upper,
|
|
'bb_lower': bb_lower,
|
|
'volatility_annual_pct': volatility.values[0] if hasattr(volatility, 'values') else volatility,
|
|
'fibonacci_levels': fib_levels
|
|
}
|
|
|
|
except Exception as e:
|
|
print(f"Error fetching technicals for {ticker}: {e}")
|
|
return None
|