File size: 10,649 Bytes
07ff2cb | 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 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 | """
Technical Analyst Tools for FinAgent.
Provides tools for technical analysis agents:
- get_price_history: Retrieve price history with calculated indicators
- calculate_indicators: Compute buy/sell signals from technical indicators
"""
# pandas-ta-remake is the maintained fork published under a different module
# name (pandas_ta_remake). Try it first, then fall back to the upstream
# pandas_ta name if it is what the environment provides.
try:
import pandas_ta_remake as ta # type: ignore
except ImportError: # pragma: no cover - exercised only when remake is absent
import pandas_ta as ta # type: ignore
import yfinance
from crewai.tools import tool
from tools.cache import TTLCache
from tools.utils import validate_ticker
cache = TTLCache()
@tool("Get Price History")
def get_price_history(ticker: str) -> str:
"""Retrieve 60 days of price history with technical indicators including RSI, MACD, SMA, and Bollinger Bands.
Args:
ticker: Stock symbol (e.g., AAPL) or crypto pair (e.g., BTC-USD).
Returns:
A formatted table with the last 5 days of price data and indicators,
or an error message if data is unavailable.
"""
try:
# 1. Input validation
valid, result = validate_ticker(ticker)
if not valid:
return result
normalized_ticker = result
# 2. Cache check
cache_key = cache.make_key("get_price_history", ticker=normalized_ticker)
cached = cache.get(cache_key)
if cached:
return cached
# 3. External API call — fetch 90 calendar days to ensure ~60 trading days
stock = yfinance.Ticker(normalized_ticker)
hist = stock.history(period="90d")
if hist is None or hist.empty:
return f"Error: Ticker '{normalized_ticker}' not found. Please verify the symbol."
close = hist["Close"]
num_days = len(close)
# 4. Calculate indicators — mark as N/A if insufficient data
insufficient_data = num_days < 50
if insufficient_data:
# Not enough data for SMA50; calculate what we can
rsi = ta.rsi(close, length=14) if num_days >= 14 else None
macd_df = ta.macd(close, fast=12, slow=26, signal=9) if num_days >= 26 else None
sma20 = ta.sma(close, length=20) if num_days >= 20 else None
sma50 = None
bbands_df = ta.bbands(close, length=20, std=2) if num_days >= 20 else None
else:
rsi = ta.rsi(close, length=14)
macd_df = ta.macd(close, fast=12, slow=26, signal=9)
sma20 = ta.sma(close, length=20)
sma50 = ta.sma(close, length=50)
bbands_df = ta.bbands(close, length=20, std=2)
# 5. Format response — last 5 days of computed data
header = (
f"Price History & Indicators for {normalized_ticker} (Last 5 Days):\n"
f"Date | Close | RSI | MACD | Signal | SMA20 | SMA50 | BB_Upper | BB_Lower"
)
rows = []
last_5_indices = hist.index[-5:]
for idx in last_5_indices:
date_str = idx.strftime("%Y-%m-%d")
close_val = f"{close[idx]:.2f}"
# RSI
if rsi is not None and idx in rsi.index and not _is_na(rsi[idx]):
rsi_val = f"{rsi[idx]:.1f}"
else:
rsi_val = "N/A"
# MACD and Signal
if macd_df is not None and idx in macd_df.index:
macd_val_raw = macd_df["MACD_12_26_9"][idx]
signal_val_raw = macd_df["MACDs_12_26_9"][idx]
macd_val = f"{macd_val_raw:.2f}" if not _is_na(macd_val_raw) else "N/A"
signal_val = f"{signal_val_raw:.2f}" if not _is_na(signal_val_raw) else "N/A"
else:
macd_val = "N/A"
signal_val = "N/A"
# SMA20
if sma20 is not None and idx in sma20.index and not _is_na(sma20[idx]):
sma20_val = f"{sma20[idx]:.2f}"
else:
sma20_val = "N/A"
# SMA50
if sma50 is not None and idx in sma50.index and not _is_na(sma50[idx]):
sma50_val = f"{sma50[idx]:.2f}"
else:
sma50_val = "N/A"
# Bollinger Bands
if bbands_df is not None and idx in bbands_df.index:
bbu_raw = bbands_df["BBU_20_2.0"][idx]
bbl_raw = bbands_df["BBL_20_2.0"][idx]
bbu_val = f"{bbu_raw:.2f}" if not _is_na(bbu_raw) else "N/A"
bbl_val = f"{bbl_raw:.2f}" if not _is_na(bbl_raw) else "N/A"
else:
bbu_val = "N/A"
bbl_val = "N/A"
row = (
f"{date_str} | {close_val:>7} | {rsi_val:>5} | {macd_val:>5} | "
f"{signal_val:>6} | {sma20_val:>7} | {sma50_val:>7} | {bbu_val:>8} | {bbl_val:>8}"
)
rows.append(row)
response = header + "\n" + "\n".join(rows)
# 6. Cache and return
cache.set(cache_key, response)
return response
except Exception as e:
return f"Error: An unexpected error occurred while processing {ticker}: {str(e)}"
def _is_na(value) -> bool:
"""Check if a value is NaN or None."""
if value is None:
return True
try:
import math
return math.isnan(value)
except (TypeError, ValueError):
return False
@tool("Calculate Indicators")
def calculate_indicators(ticker: str) -> str:
"""Compute current buy/sell signals from RSI, MACD, and Bollinger Bands for a given ticker.
Args:
ticker: Stock symbol (e.g., AAPL) or crypto pair (e.g., BTC-USD).
Returns:
A formatted string with each indicator's current value and signal classification
(BUY, SELL, or NEUTRAL), or an error message if data is unavailable.
"""
try:
# 1. Input validation
valid, result = validate_ticker(ticker)
if not valid:
return result
normalized_ticker = result
# 2. Cache check
cache_key = cache.make_key("calculate_indicators", ticker=normalized_ticker)
cached = cache.get(cache_key)
if cached:
return cached
# 3. External API call — fetch 90 calendar days to ensure ~60 trading days
stock = yfinance.Ticker(normalized_ticker)
hist = stock.history(period="90d")
if hist is None or hist.empty:
return f"Error: Ticker '{normalized_ticker}' not found. Please verify the symbol."
close = hist["Close"]
num_days = len(close)
if num_days < 26:
return f"Error: Insufficient data for {normalized_ticker}. Need at least 26 trading days."
# 4. Calculate indicators
rsi_series = ta.rsi(close, length=14)
macd_df = ta.macd(close, fast=12, slow=26, signal=9)
bbands_df = ta.bbands(close, length=20, std=2)
# Get current values
current_rsi = rsi_series.iloc[-1] if rsi_series is not None else None
current_close = close.iloc[-1]
# MACD values
if macd_df is not None:
current_macd = macd_df["MACD_12_26_9"].iloc[-1]
current_signal = macd_df["MACDs_12_26_9"].iloc[-1]
prev_macd = macd_df["MACD_12_26_9"].iloc[-2]
prev_signal = macd_df["MACDs_12_26_9"].iloc[-2]
else:
current_macd = None
current_signal = None
prev_macd = None
prev_signal = None
# Bollinger Band values
if bbands_df is not None:
current_upper = bbands_df["BBU_20_2.0"].iloc[-1]
current_lower = bbands_df["BBL_20_2.0"].iloc[-1]
else:
current_upper = None
current_lower = None
# 5. Classify signals
# RSI classification
if current_rsi is not None and not _is_na(current_rsi):
if current_rsi < 30:
rsi_signal = "BUY"
rsi_desc = "Oversold"
elif current_rsi > 70:
rsi_signal = "SELL"
rsi_desc = "Overbought"
else:
rsi_signal = "NEUTRAL"
rsi_desc = "Neutral"
rsi_line = f"RSI (14): {current_rsi:.1f} → {rsi_signal} ({rsi_desc})"
else:
rsi_line = "RSI (14): N/A → NEUTRAL (Insufficient Data)"
# MACD classification
if (current_macd is not None and current_signal is not None and
prev_macd is not None and prev_signal is not None and
not _is_na(current_macd) and not _is_na(current_signal) and
not _is_na(prev_macd) and not _is_na(prev_signal)):
is_bullish = current_macd > current_signal and prev_macd <= prev_signal
is_bearish = current_macd < current_signal and prev_macd >= prev_signal
if is_bullish:
macd_signal = "BUY"
macd_desc = "Bullish Crossover"
elif is_bearish:
macd_signal = "SELL"
macd_desc = "Bearish Crossover"
else:
macd_signal = "NEUTRAL"
macd_desc = "Neutral"
macd_line = f"MACD: {current_macd:.2f} / Signal: {current_signal:.2f} → {macd_signal} ({macd_desc})"
else:
macd_line = "MACD: N/A / Signal: N/A → NEUTRAL (Insufficient Data)"
# Bollinger classification
if (current_upper is not None and current_lower is not None and
not _is_na(current_upper) and not _is_na(current_lower)):
if current_close < current_lower:
bb_signal = "BUY"
bb_desc = "Below Lower Band"
elif current_close > current_upper:
bb_signal = "SELL"
bb_desc = "Above Upper Band"
else:
bb_signal = "NEUTRAL"
bb_desc = "Neutral"
bb_line = f"Bollinger: Price ${current_close:.2f} / Upper ${current_upper:.2f} / Lower ${current_lower:.2f} → {bb_signal} ({bb_desc})"
else:
bb_line = "Bollinger: N/A → NEUTRAL (Insufficient Data)"
# 6. Format response
response = (
f"Technical Signals for {normalized_ticker}:\n"
f"{rsi_line}\n"
f"{macd_line}\n"
f"{bb_line}"
)
# 7. Cache and return
cache.set(cache_key, response)
return response
except Exception as e:
return f"Error: An unexpected error occurred while processing {ticker}: {str(e)}"
|