Download src/crypto_analysis/tools/technical_tools.py from vlbandara/cryptic: direct link, hf CLI and curl.
- Browser
- Download file 17.8 kB
-
https://huggingface.co/spaces/vlbandara/cryptic/resolve/main/src/crypto_analysis/tools/technical_tools.py
- Command line
-
hf download hf://spaces/vlbandara/cryptic/src/crypto_analysis/tools/technical_tools.py
-
curl -L -o technical_tools.py https://huggingface.co/spaces/vlbandara/cryptic/resolve/main/src/crypto_analysis/tools/technical_tools.py
17.8 kB
| import os | |
| import requests | |
| import json | |
| from typing import Dict, Any, List, Optional | |
| from datetime import datetime, timedelta | |
| import pandas as pd | |
| import numpy as np | |
| import pytz # Import pytz for timezone handling | |
| # Patch for pandas_ta NaN import issue | |
| import sys | |
| import importlib.util | |
| import inspect | |
| # Create a patched version of pandas_ta that won't try to import NaN from numpy | |
| try: | |
| import pandas_ta as ta | |
| except ImportError as e: | |
| if "cannot import name 'NaN' from 'numpy'" in str(e): | |
| # Fix by monkeypatching numpy to provide NaN in expected location | |
| np.NaN = np.nan | |
| # Now import pandas_ta | |
| import pandas_ta as ta | |
| else: | |
| raise | |
| # Import OpenAI for LLM strategy interpretation | |
| import openai | |
| from dotenv import load_dotenv | |
| load_dotenv() | |
| from crewai.tools import BaseTool | |
| from pydantic import Field | |
| from alpaca.data.historical import CryptoHistoricalDataClient | |
| from alpaca.data.requests import CryptoBarsRequest | |
| from alpaca.data.timeframe import TimeFrame, TimeFrameUnit | |
| # Set up OpenAI API key | |
| openai.api_key = os.getenv("OPENAI_API_KEY") | |
| class IndicatorCalculator: | |
| """A modular class to calculate various technical indicators on price data.""" | |
| # Dictionary of available indicators with their descriptions | |
| AVAILABLE_INDICATORS = { | |
| "rsi": "Relative Strength Index - measures momentum by comparing recent gains to recent losses", | |
| "bollinger_bands": "Bollinger Bands - measure volatility with upper and lower bands", | |
| "macd": "Moving Average Convergence Divergence - momentum indicator showing relationship between two moving averages", | |
| "adx": "Average Directional Index - measures trend strength", | |
| "ema": "Exponential Moving Average - gives more weight to recent prices", | |
| "sma": "Simple Moving Average - arithmetic mean of prices over a period", | |
| "atr": "Average True Range - measures volatility", | |
| "stochastic": "Stochastic Oscillator - momentum indicator comparing close price to price range", | |
| "obv": "On-Balance Volume - relates volume to price change" | |
| } | |
| def __init__(self): | |
| """Initialize the indicator calculator.""" | |
| pass | |
| def calculate_all_indicators(self, df: pd.DataFrame, params: Dict[str, Any] = None) -> pd.DataFrame: | |
| """ | |
| Calculate all available indicators on the price data | |
| Args: | |
| df: DataFrame with OHLCV data | |
| params: Dictionary of parameters for indicators | |
| Returns: | |
| DataFrame with added technical indicators | |
| """ | |
| if params is None: | |
| params = {} | |
| try: | |
| # Ensure we're working with a DataFrame with expected columns | |
| required_columns = ['open', 'high', 'low', 'close', 'volume'] | |
| missing_columns = [col for col in required_columns if col not in df.columns] | |
| if missing_columns: | |
| print(f"Missing columns in dataframe: {missing_columns}") | |
| print(f"Available columns: {df.columns}") | |
| return df | |
| # Calculate RSI | |
| rsi_length = params.get('rsi_length', 14) | |
| df = self.calculate_rsi(df, length=rsi_length) | |
| # Calculate Bollinger Bands | |
| bb_length = params.get('bb_length', 20) | |
| bb_std = params.get('bb_std', 2.0) | |
| df = self.calculate_bollinger_bands(df, length=bb_length, std=bb_std) | |
| # Calculate ADX | |
| adx_length = params.get('adx_length', 14) | |
| df = self.calculate_adx(df, length=adx_length) | |
| # Calculate EMAs | |
| ema_lengths = params.get('ema_lengths', [8, 21, 50, 200]) | |
| for length in ema_lengths: | |
| df = self.calculate_ema(df, length=length) | |
| # Calculate MACD | |
| macd_fast = params.get('macd_fast', 12) | |
| macd_slow = params.get('macd_slow', 26) | |
| macd_signal = params.get('macd_signal', 9) | |
| df = self.calculate_macd(df, fast=macd_fast, slow=macd_slow, signal=macd_signal) | |
| # Calculate Stochastic Oscillator | |
| stoch_k = params.get('stoch_k', 14) | |
| stoch_d = params.get('stoch_d', 3) | |
| df = self.calculate_stochastic(df, k=stoch_k, d=stoch_d) | |
| # Calculate ATR | |
| atr_length = params.get('atr_length', 14) | |
| df = self.calculate_atr(df, length=atr_length) | |
| # Calculate OBV | |
| df = self.calculate_obv(df) | |
| print(f"Calculated all technical indicators. Final data shape: {df.shape}") | |
| return df | |
| except Exception as e: | |
| print(f"Error calculating indicators: {e}") | |
| import traceback | |
| traceback.print_exc() | |
| return df | |
| def calculate_rsi(self, df: pd.DataFrame, length: int = 14) -> pd.DataFrame: | |
| """Calculate RSI indicator.""" | |
| df[f'rsi_{length}'] = ta.rsi(df['close'], length=length) | |
| return df | |
| def calculate_bollinger_bands(self, df: pd.DataFrame, length: int = 20, std: float = 2.0) -> pd.DataFrame: | |
| """Calculate Bollinger Bands indicator.""" | |
| bbands = ta.bbands(df['close'], length=length, std=std) | |
| df[f'bb_upper_{length}_{std}'] = bbands[f'BBU_{length}_{std}'] | |
| df[f'bb_middle_{length}_{std}'] = bbands[f'BBM_{length}_{std}'] | |
| df[f'bb_lower_{length}_{std}'] = bbands[f'BBL_{length}_{std}'] | |
| # Normalize Bollinger Bands position (0 = lower band, 1 = upper band) | |
| df[f'bb_position_{length}_{std}'] = (df['close'] - df[f'bb_lower_{length}_{std}']) / (df[f'bb_upper_{length}_{std}'] - df[f'bb_lower_{length}_{std}']) | |
| return df | |
| def calculate_adx(self, df: pd.DataFrame, length: int = 14) -> pd.DataFrame: | |
| """Calculate ADX indicator.""" | |
| adx = ta.adx(df['high'], df['low'], df['close'], length=length) | |
| df[f'adx_{length}'] = adx[f'ADX_{length}'] | |
| df[f'di_plus_{length}'] = adx[f'DMP_{length}'] | |
| df[f'di_minus_{length}'] = adx[f'DMN_{length}'] | |
| return df | |
| def calculate_ema(self, df: pd.DataFrame, length: int = 21) -> pd.DataFrame: | |
| """Calculate EMA indicator.""" | |
| df[f'ema_{length}'] = ta.ema(df['close'], length=length) | |
| return df | |
| def calculate_sma(self, df: pd.DataFrame, length: int = 21) -> pd.DataFrame: | |
| """Calculate SMA indicator.""" | |
| df[f'sma_{length}'] = ta.sma(df['close'], length=length) | |
| return df | |
| def calculate_macd(self, df: pd.DataFrame, fast: int = 12, slow: int = 26, signal: int = 9) -> pd.DataFrame: | |
| """Calculate MACD indicator.""" | |
| macd = ta.macd(df['close'], fast=fast, slow=slow, signal=signal) | |
| df[f'macd_{fast}_{slow}_{signal}'] = macd[f'MACD_{fast}_{slow}_{signal}'] | |
| df[f'macd_signal_{fast}_{slow}_{signal}'] = macd[f'MACDs_{fast}_{slow}_{signal}'] | |
| df[f'macd_histogram_{fast}_{slow}_{signal}'] = macd[f'MACDh_{fast}_{slow}_{signal}'] | |
| return df | |
| def calculate_stochastic(self, df: pd.DataFrame, k: int = 14, d: int = 3) -> pd.DataFrame: | |
| """Calculate Stochastic Oscillator.""" | |
| stoch = ta.stoch(df['high'], df['low'], df['close'], k=k, d=d) | |
| df[f'stoch_k_{k}'] = stoch[f'STOCHk_{k}_{d}_3'] | |
| df[f'stoch_d_{k}_{d}'] = stoch[f'STOCHd_{k}_{d}_3'] | |
| return df | |
| def calculate_atr(self, df: pd.DataFrame, length: int = 14) -> pd.DataFrame: | |
| """Calculate Average True Range.""" | |
| df[f'atr_{length}'] = ta.atr(df['high'], df['low'], df['close'], length=length) | |
| return df | |
| def calculate_obv(self, df: pd.DataFrame) -> pd.DataFrame: | |
| """Calculate On-Balance Volume.""" | |
| df['obv'] = ta.obv(df['close'], df['volume']) | |
| return df | |
| def get_available_indicators(cls) -> Dict[str, str]: | |
| """Return the dictionary of available indicators with descriptions.""" | |
| return cls.AVAILABLE_INDICATORS | |
| class TechnicalAnalysisStrategy(BaseTool): | |
| name: str = "Bitcoin Technical Analysis Strategy Tool" | |
| description: str = "Fetches current Bitcoin technical indicators data from the market" | |
| # Define all fields as proper Pydantic fields | |
| api_key: Optional[str] = Field(default=None, description="Alpaca API key") | |
| api_secret: Optional[str] = Field(default=None, description="Alpaca API secret") | |
| client: Optional[Any] = Field(default=None, description="Alpaca client instance") | |
| indicator_calculator: IndicatorCalculator = Field(default_factory=IndicatorCalculator, description="Indicator calculator instance") | |
| # Add model_config to allow arbitrary types | |
| model_config = {"arbitrary_types_allowed": True} | |
| def __init__(self, **kwargs): | |
| # Initialize the base class first | |
| super().__init__(**kwargs) | |
| # Set the API keys from environment variables if not provided directly | |
| if not self.api_key: | |
| self.api_key = os.getenv("ALPACA_API_KEY") | |
| if not self.api_secret: | |
| self.api_secret = os.getenv("ALPACA_API_SECRET") | |
| print(f"Initializing Technical Analysis Strategy with API key: {'Present' if self.api_key else 'Missing'}") | |
| # Initialize the Alpaca client if not already set | |
| if not self.client and self.api_key and self.api_secret: | |
| try: | |
| self.client = CryptoHistoricalDataClient(api_key=self.api_key, secret_key=self.api_secret) | |
| print("Successfully initialized Alpaca client") | |
| except Exception as e: | |
| print(f"Error initializing Alpaca client: {e}") | |
| def _run(self) -> Dict[str, Any]: | |
| """ | |
| Fetch Bitcoin technical indicator data | |
| Returns: | |
| Dictionary with all technical indicators data | |
| """ | |
| try: | |
| print("Fetching Bitcoin technical indicator data") | |
| # Make sure client is initialized | |
| if not self.client: | |
| print("Client not initialized, attempting to create one now") | |
| self.client = CryptoHistoricalDataClient(api_key=self.api_key, secret_key=self.api_secret) | |
| # Request extra periods to account for gaps in historical data | |
| # Request 2x the minimum to ensure we get enough data points | |
| min_required = 20 # Require fewer data points for basic analysis | |
| lookback_periods = 100 | |
| actual_lookback = max(min_required, lookback_periods) * 3 | |
| timeframe_minutes = 5 # Default timeframe | |
| # Fetch the Bitcoin price data for the specified timeframe | |
| df = self._fetch_price_data(actual_lookback, timeframe_minutes) | |
| if df is None: | |
| print("No price data returned from fetch_price_data") | |
| return {"error": "No price data available"} | |
| print(f"Retrieved {len(df)} data points") | |
| if len(df) < min_required: | |
| print(f"Insufficient data points: {len(df)}, need at least {min_required}") | |
| return {"error": f"Insufficient price data: only received {len(df)} data points"} | |
| # Calculate all indicators | |
| indicator_params = { | |
| 'bb_std': 2.0, | |
| 'rsi_length': 14, | |
| 'bb_length': 20 | |
| } | |
| df = self.indicator_calculator.calculate_all_indicators(df, indicator_params) | |
| # Get the latest data point with all indicators | |
| latest_data = df.iloc[-1].to_dict() | |
| # Prepare result dictionary | |
| result = {} | |
| # Add price data | |
| result["price"] = float(latest_data.get('close', 0)) | |
| # Add all indicators | |
| for key, value in latest_data.items(): | |
| if key not in ['open', 'high', 'low', 'close', 'volume', 'time']: | |
| # Convert numpy values to Python native types | |
| if pd.isna(value): | |
| result[key] = None | |
| else: | |
| result[key] = float(value) | |
| return result | |
| except Exception as e: | |
| print(f"Error fetching indicator data: {str(e)}") | |
| return {"error": str(e)} | |
| def _fetch_price_data(self, lookback_periods: int, timeframe_minutes: int = 5) -> pd.DataFrame: | |
| """ | |
| Fetch Bitcoin price data from Alpaca API | |
| Args: | |
| lookback_periods: Number of periods to fetch | |
| timeframe_minutes: Timeframe in minutes | |
| Returns: | |
| DataFrame with OHLCV data | |
| """ | |
| try: | |
| # Calculate the start and end dates with timezone information | |
| end = datetime.now(pytz.UTC) # Make end timezone-aware with UTC | |
| start = end - timedelta(minutes=timeframe_minutes * lookback_periods) | |
| print(f"Fetching price data from {start} to {end}") | |
| # Create the request parameters | |
| request_params = CryptoBarsRequest( | |
| symbol_or_symbols=["BTC/USD"], # Use correct format with slash | |
| timeframe=TimeFrame(timeframe_minutes, TimeFrameUnit.Minute), # Use specified timeframe | |
| start=start, | |
| end=end | |
| ) | |
| print(f"Request parameters: {request_params}") | |
| # Get the bars data | |
| bars = self.client.get_crypto_bars(request_params) | |
| if bars is None: | |
| print("No bars data returned from API") | |
| return None | |
| # Convert to dataframe | |
| df = bars.df.reset_index() | |
| print(f"Raw data columns: {df.columns}") | |
| print(f"Raw data shape: {df.shape}") | |
| # Print first few rows for debugging | |
| print(f"First few rows: {df.head(2)}") | |
| # Ensure proper column names | |
| if 'timestamp' in df.columns: | |
| df = df.rename(columns={'timestamp': 'time'}) | |
| # Filter to only BTC/USD data | |
| if 'symbol' in df.columns: | |
| df = df[df['symbol'] == 'BTC/USD'].reset_index(drop=True) | |
| df = df.drop(columns=['symbol']) | |
| print(f"Processed data shape: {df.shape}") | |
| return df | |
| except Exception as e: | |
| print(f"Error fetching price data with SDK: {e}") | |
| print(f"API Key present: {'Yes' if self.api_key else 'No'}") | |
| print(f"API Secret present: {'Yes' if self.api_secret else 'No'}") | |
| # Try the direct REST API approach as fallback | |
| try: | |
| print("Attempting fallback to direct REST API call...") | |
| return self._fetch_price_data_direct_api(lookback_periods, timeframe_minutes) | |
| except Exception as e2: | |
| print(f"Error in fallback API call: {e2}") | |
| return None | |
| def _fetch_price_data_direct_api(self, lookback_periods: int, timeframe_minutes: int = 5) -> pd.DataFrame: | |
| """ | |
| Fallback method to fetch Bitcoin price data using direct REST API calls | |
| following the curl example format. | |
| """ | |
| # Calculate the start and end dates | |
| end = datetime.now(pytz.UTC) | |
| start = end - timedelta(minutes=timeframe_minutes * lookback_periods) | |
| # Format dates for the API | |
| start_str = start.strftime('%Y-%m-%dT%H:%M:%SZ') | |
| end_str = end.strftime('%Y-%m-%dT%H:%M:%SZ') | |
| # Endpoint for historical bars | |
| url = f"https://data.alpaca.markets/v1beta3/crypto/us/bars" | |
| # Query parameters | |
| params = { | |
| "symbols": "BTC/USD", | |
| "timeframe": f"{timeframe_minutes}Min", | |
| "start": start_str, | |
| "end": end_str, | |
| "limit": lookback_periods | |
| } | |
| # Headers with authentication | |
| headers = { | |
| "Apca-Api-Key-Id": self.api_key, | |
| "Apca-Api-Secret-Key": self.api_secret | |
| } | |
| print(f"Making direct API call to: {url}") | |
| print(f"With params: {params}") | |
| # Make the request | |
| response = requests.get(url, params=params, headers=headers) | |
| if response.status_code != 200: | |
| print(f"API Error: {response.status_code} - {response.text}") | |
| return None | |
| # Parse response | |
| data = response.json() | |
| print(f"API Response: {json.dumps(data)[:300]}...") | |
| # Extract the bars | |
| if 'bars' not in data or 'BTC/USD' not in data['bars']: | |
| print("No bar data found in response") | |
| return None | |
| bars_data = data['bars']['BTC/USD'] | |
| # Create DataFrame | |
| df = pd.DataFrame(bars_data) | |
| # Rename columns to match expected format | |
| if 't' in df.columns: | |
| df = df.rename(columns={ | |
| 't': 'time', | |
| 'o': 'open', | |
| 'h': 'high', | |
| 'l': 'low', | |
| 'c': 'close', | |
| 'v': 'volume' | |
| }) | |
| # Convert timestamp to datetime | |
| if 'time' in df.columns: | |
| df['time'] = pd.to_datetime(df['time']) | |
| print(f"Processed API data shape: {df.shape}") | |
| return df |