File: //opt/af/scripts/moreagents.txt
B. Adapting Forex Logic for Agno Agents
1. Input Handling
The Forex strategies identified require specific data inputs. Prediction agents like "FXSeer LSTM" or "TrendSight TA" will need historical OHLCV data for one or more currency pairs. This data would likely be fetched via the data_fetcher tool module, either during initialization or periodically within the agent's execution cycle. Trading idea agents like "Momentum Mover" or "Mean Reversion Scout" might also fetch historical or near-real-time data, or they could be designed to consume the output of prediction agents. Defining standardized input formats is crucial. For instance, historical data could consistently be passed as Pandas DataFrames, while predictions from one agent to another could use a dictionary or JSON structure containing key information like asset, timestamp, predicted value/direction, and confidence.
2. Output Handling
Similarly, the outputs of the agents must be structured consistently.
Prediction Agents (e.g., FXSeer LSTM, TrendSight TA): Should produce outputs clearly indicating the asset, the prediction horizon (e.g., next day, next bar), the prediction itself (e.g., specific price, direction UP/DOWN/SIDEWAYS, condition OVERBOUGHT/OVERSOLD), and potentially a confidence score or the basis for the prediction (e.g., "RSI > 70"). A dictionary or JSON format is suitable.
Trading Idea Agents (e.g., Momentum Mover, Mean Reversion Scout): Should generate actionable ideas. The output should specify the asset, the suggested action (BUY/SELL/HOLD), potentially an indicative entry price, stop-loss/take-profit levels if derivable from the strategy, the rationale (e.g., "Top Momentum Rank", "Z-score < -2"), and a confidence level. Again, a structured format like a dictionary or JSON object (or a list of them for agents generating multiple ideas) is appropriate.
3. State Management
Agents may need to maintain state between execution cycles. For example, an LSTM agent needs its loaded model, a TA agent might need the previous period's indicator values to detect crossovers, and a mean reversion agent needs its calculated moving average or bands. This state would typically be stored as attributes of the agent's class instance.
The successful integration of Forex logic into the Agno framework hinges on carefully designing these interfaces. How an agent receives data, how it processes that data using its core strategy logic, and how it formats its output for the framework or other agents are critical design considerations. Standardizing these input/output formats using structures like dictionaries or JSON objects is essential for enabling potential collaboration and communication between different agents within the Agno system (e.g., allowing "Momentum Mover" to optionally use predictions from "TrendSight TA"). Clear interfaces prevent integration issues and facilitate the creation of more complex, multi-agent systems.
IV. Forex Prediction Agents for Agno
A. Agent 1: "FXSeer LSTM" (LSTM Predictor)
1. Marketing Name Rationale
The name "FXSeer LSTM" combines "FX" for Forex, "Seer" to imply predictive capability, and "LSTM" to clearly identify the core machine learning technique employed.
2. Approach
This agent implements a Long Short-Term Memory (LSTM) neural network approach for time series forecasting, drawing inspiration from examples that predict stock prices using historical data. The fundamental idea is that LSTMs, being a type of recurrent neural network, are well-suited for learning patterns in sequential data like financial price series. The agent will use a sequence of past Forex price data points (e.g., daily adjusted closing prices for EUR/USD over the last 20 days) to predict the price for the next period (e.g., the price on day 21).
3. Free Inputs
Data: Historical Open, High, Low, Close, Volume (OHLCV) data for a specific Forex pair (e.g., EUR/USD). This data will be sourced via the data_fetcher tool, utilizing free API tiers like Alpha Vantage or yfinance.
Configuration:
Currency pair (e.g., 'EUR/USD').
Data interval/frequency (e.g., 'Daily').
Lookback period (number of past periods used for prediction, e.g., 20 ).
Prediction horizon (number of periods ahead to predict, e.g., 1).
Path to a pre-trained LSTM model file.
Path to the scaler object used during training (for data normalization/denormalization).
API Key (if using Alpha Vantage).
4. Core Logic
* On initialization or during its run cycle, the agent requests the required historical data (lookback period + recent data) for the specified currency pair using the data_fetcher tool.
* The relevant feature (e.g., 'Adjusted Close' or 'Close') is selected from the fetched DataFrame.
* Data normalization is applied using a pre-fitted scaler (e.g., MinMaxScaler from scikit-learn), loaded from the specified path. Normalization is crucial for improving LSTM training convergence and performance.
* A pre-trained LSTM model (developed using PyTorch or TensorFlow ) is loaded using the lstm_model_handler tool. Note: The process of training this model is considered separate from the agent's runtime operation. A script for training could be provided alongside the agent code.
* The most recent sequence of normalized data, matching the lookback period, is formatted into the shape expected by the LSTM model's input layer.
* The lstm_model_handler is used to pass the input sequence through the loaded model to generate a normalized prediction.
* The predicted value is inverse-transformed using the loaded scaler to bring it back to the original price scale.
* The final prediction is formatted into the standard output structure.
5. Output
The agent produces a dictionary or JSON object containing the prediction details:
JSON
{
"agent_name": "FXSeer LSTM",
"timestamp": "2024-05-15T10:00:00Z",
"asset": "EUR/USD",
"prediction_horizon": "1D",
"predicted_price": 1.0850,
"confidence": 0.75 // Optional: Could be derived from model uncertainty if available
}
6. Required Tools
data_fetcher.py: To acquire historical Forex data.
lstm_model_handler.py: To load the pre-trained model and scaler, and execute the prediction step.
External Libraries: pandas, numpy, scikit-learn (for scaling), pytorch or tensorflow (matching the framework used for the pre-trained model).
B. Agent 2: "TrendSight TA" (TA-based Predictor)
1. Marketing Name Rationale
"TrendSight TA" suggests gaining insights ("Sight") from market trends ("Trend") using Technical Analysis ("TA").
2. Approach
This agent leverages technical indicators to forecast the likely short-term price direction (e.g., Up, Down, Sideways) or identify market conditions (e.g., Overbought, Oversold). It uses standard TA libraries like pandas-ta or ta to compute indicators such as the Relative Strength Index (RSI), Moving Average Convergence Divergence (MACD), or Bollinger Bands. The prediction logic is based on predefined rules applied to the latest indicator values, such as:
Threshold Crossing: e.g., RSI moving above 70 suggests an overbought condition (potential Down move), while RSI below 30 suggests oversold (potential Up move).
Indicator Crossovers: e.g., The MACD line crossing above the MACD signal line indicates bullish momentum (potential Up move), while crossing below suggests bearish momentum (potential Down move).
Band Interaction: e.g., Price closing above the upper Bollinger Band might indicate strong momentum or an overbought condition, while closing below the lower band suggests strong negative momentum or an oversold condition.
3. Free Inputs
Data: Historical OHLCV data for a specific Forex pair (e.g., GBP/JPY) obtained via data_fetcher using free APIs.
Configuration:
Currency pair (e.g., 'GBP/JPY').
Data interval (e.g., '1h', '4h').
Indicator configurations (e.g., { 'indicator': 'RSI', 'period': 14, 'threshold_high': 70, 'threshold_low': 30 }, { 'indicator': 'MACD', 'fast': 12, 'slow': 26, 'signal': 9 }).
Prediction logic identifier (e.g., "RSI_threshold", "MACD_crossover", "BBands_break").
API Key/Source identifier.
4. Core Logic
* Fetch the required amount of historical data using data_fetcher.
* Utilize the technical_analyzer.py tool module to calculate the technical indicator(s) specified in the configuration, adding them as columns to the DataFrame.
* Apply the selected prediction logic based on the configuration:
* For threshold logic: Check if the latest indicator value crosses the defined high or low threshold.
* For crossover logic: Compare the latest values of the two relevant indicator lines (e.g., MACD line vs. signal line) and their previous values to detect a cross.
* For band logic: Check if the latest closing price is above the upper band or below the lower band.
* Determine the predicted direction (UP/DOWN/SIDEWAYS) or condition (OVERBOUGHT/OVERSOLD) based on the outcome of the logic check.
* Optionally assign a signal strength (e.g., Strong, Moderate, Weak) based on the magnitude of the threshold break or other factors.
* Format the prediction into the standard output structure.
5. Output
The agent produces a dictionary or JSON object detailing the TA-based prediction:
JSON
{
"agent_name": "TrendSight TA",
"timestamp": "2024-05-15T11:00:00Z",
"asset": "GBP/JPY",
"prediction_horizon": "Next Bar", // Or specific timeframe like '1h'
"predicted_direction": "UP", // Or DOWN, SIDEWAYS, OVERBOUGHT, OVERSOLD
"signal_strength": "Moderate", // Optional
"basis": "MACD Crossover Bullish" // Explanation of the signal trigger
}
6. Required Tools
data_fetcher.py: For obtaining historical Forex data.
technical_analyzer.py: For calculating the required technical indicators.
External Libraries: pandas.
V. Forex Trading Idea Agents for Agno
A. Agent 1: "Momentum Mover" (Momentum Strategy)
1. Marketing Name Rationale
"Momentum" clearly states the strategy type, while "Mover" implies taking action based on observed market momentum.
2. Approach
This agent implements a relative strength momentum strategy, inspired by concepts discussed in quantitative trading resources. The core idea is that assets that have performed well recently tend to continue performing well in the short term (and vice-versa). The agent calculates the percentage price change (momentum) over a defined lookback period (e.g., 1 month, 3 months) for a predefined list of Forex pairs. It then ranks the pairs based on this momentum. A "BUY" trading idea is generated for the pair(s) exhibiting the strongest positive momentum, and potentially a "SELL" idea for the pair(s) with the strongest negative momentum (weakest performance). This is a simplified adaptation, focusing on signal generation rather than the complex portfolio construction and weighting often found in full momentum strategies.
3. Free Inputs
Data: Historical price data (e.g., daily closing prices) for a list of Forex pairs (e.g., ``). Sourced via data_fetcher.
Configuration:
List of Forex pairs to analyze.
Momentum calculation period (e.g., '1M', '3M', '20D').
Number of top performers to generate "BUY" ideas for (e.g., 1, 2).
Number of bottom performers to generate "SELL" ideas for (e.g., 1, 2).
API Key/Source identifier.
Optional: Input source for predictions from other agents (e.g., to filter momentum signals).
4. Core Logic
* Fetch historical price data for all specified currency pairs using data_fetcher, ensuring sufficient history for the momentum calculation period.
* For each pair, calculate the percentage return (momentum) over the configured lookback period (e.g., (last_price / price_n_periods_ago) - 1).
* Rank the pairs in descending order based on their calculated momentum.
* Identify the top N pairs (strongest positive momentum) and the bottom N pairs (strongest negative momentum or lowest positive momentum).
* Generate a "BUY" trading idea dictionary for each of the top N pairs.
* Generate a "SELL" trading idea dictionary for each of the bottom N pairs.
* Optionally, filter these ideas based on predictions received from other agents (if configured).
* Compile the generated ideas into a list.
5. Output
The agent produces a list of dictionaries/JSON objects, each representing a trading idea:
JSON
6. Required Tools
data_fetcher.py: To fetch data for multiple Forex pairs.
External Libraries: pandas, numpy.
B. Agent 2: "Mean Reversion Scout" (Mean Reversion Strategy)
1. Marketing Name Rationale
"Mean Reversion" identifies the core trading philosophy (prices tend to revert to their average), while "Scout" suggests the agent is searching for these reversion opportunities.
2. Approach
This agent seeks opportunities based on the principle of mean reversion, where prices that have moved significantly away from their recent average are expected to move back towards it. Inspiration comes from strategies using Z-scores or Bollinger Bands. The agent calculates a measure of price deviation from a central tendency (e.g., a simple moving average). This could be:
Z-Score: Calculating how many standard deviations the current price is away from a moving average ((price - moving_average) / moving_standard_deviation).
Bollinger Bands: Comparing the current price to the upper and lower bands, which are typically set at a certain number of standard deviations (e.g., 2) away from a central moving average. A "BUY" idea is generated when the price is deemed significantly below the mean (oversold condition, e.g., Z-score < -2 or price below lower Bollinger Band). A "SELL" idea is generated when the price is significantly above the mean (overbought condition, e.g., Z-score > +2 or price above upper Bollinger Band).
3. Free Inputs
Data: Historical OHLCV data for a specific Forex pair (e.g., USD/CAD) via data_fetcher.
Configuration:
Currency pair (e.g., 'USD/CAD').
Data interval (e.g., '1h').
Lookback period for calculating the moving average and standard deviation (e.g., 20).
Threshold for signal generation (e.g., Z-score absolute value 2.0, or Bollinger Band standard deviation 2.0).
Method ('ZScore' or 'BollingerBands').
API Key/Source identifier.
Optional: Input source for predictions from other agents (e.g., confirming divergence or expected reversion).
4. Core Logic
* Fetch the required historical data using data_fetcher.
* Based on the configured method:
* Z-Score: Calculate the moving average and moving standard deviation of the closing price over the lookback period. Compute the Z-score for the most recent price point.
* Bollinger Bands: Use the technical_analyzer.py tool to calculate the upper band, middle band (moving average), and lower band.
* Compare the Z-score or the price's position relative to the Bollinger Bands against the configured threshold(s).
* If the oversold condition is met (e.g., Z-score < negative threshold, or price < lower band), generate a "BUY" trading idea dictionary. Include the current price as a potential entry point.
* If the overbought condition is met (e.g., Z-score > positive threshold, or price > upper band), generate a "SELL" trading idea dictionary. Include the current price as a potential entry point.
* Optionally, filter ideas based on external predictions if configured.
* Format the output.
5. Output
The agent produces a dictionary or JSON object representing a single trading idea when a condition is met:
JSON
{
"agent_name": "Mean Reversion Scout",
"timestamp": "2024-05-15T13:00:00Z",
"asset": "USD/CAD",
"idea_type": "BUY", // Or SELL
"rationale": "Z-score < -2.0", // Or "Price below Lower BBand"
"suggested_entry": 1.3510, // Current price at signal time
"confidence": 0.7 // Optional: Could relate to magnitude of Z-score/band break
}
(Note: If no threshold is met, the agent might output nothing or a 'HOLD' signal).
6. Required Tools
data_fetcher.py: For fetching historical data.
technical_analyzer.py: Required if using the Bollinger Bands method.
External Libraries: pandas, numpy, potentially scipy.stats for Z-score calculation (though easily done with pandas/numpy).
Offering agents based on both Momentum and Mean Reversion provides users with tools reflecting distinct market hypotheses. Momentum strategies aim to capitalize on established trends, while mean reversion strategies bet on the reversal of short-term extremes. Financial markets often exhibit characteristics suitable for both approaches at different times. The relative simplicity of calculating momentum (returns) and deviation (Z-scores or Bollinger Bands) makes these strategies feasible to implement using the basic price data available from free APIs and standard calculations offered by libraries like pandas, numpy, and TA libraries , rendering them suitable candidates for this project focused on free resources.
VI. Modular Python Implementation
This section outlines the structure and core functionality of the Python modules required for the Agno agents. The code provided is illustrative and structural, designed to fit within the 150-250 LoC guideline per file and requiring adaptation for the specific Agno framework integration and error handling robustness.
A. Core Tool Modules
1. data_fetcher.py
Purpose: Centralizes API calls to fetch Forex data, abstracting the specific source (Alpha Vantage, yfinance) from the agents.
Functionality: Provides functions to get historical OHLCV data. Includes basic error handling for API requests and potential rate limit management (e.g., using time.sleep). API keys should be handled securely (e.g., via environment variables or a config file). Caching results locally (e.g., to disk) can significantly reduce API calls and improve performance.
Libraries: alpha_vantage, yfinance, pandas, os, time, logging
Python
# data_fetcher.py (Conceptual Structure)
import os
import time
import logging
import pandas as pd
from alpha_vantage.foreignexchange import ForeignExchange
from alpha_vantage.timeseries import TimeSeries # If fetching FX via stock endpoint variation
import yfinance as yf
logging.basicConfig(level=logging.INFO)
# --- Constants ---
CACHE_DIR = "data_cache"
CACHE_EXPIRY_SECONDS = 3600 # 1 hour
# --- Helper Functions ---
def _load_from_cache(symbol, interval):
#... implementation to load data if fresh cache exists...
pass # Returns DataFrame or None
def _save_to_cache(df, symbol, interval):
#... implementation to save DataFrame to cache...
pass
# --- Alpha Vantage Fetching ---
def get_fx_data_alphavantage(api_key, from_symbol, to_symbol, interval='60min', outputsize='compact'):
"""Fetches FX data from Alpha Vantage with basic caching."""
# Note: Alpha Vantage FX endpoints might differ slightly; adjust as needed.
# Using TimeSeries endpoint for daily adjusted as an example structure
symbol_pair = f"{from_symbol}{to_symbol}"
cache_key = f"AV_{symbol_pair}_{interval}"
cached_df = _load_from_cache(cache_key, interval)
if cached_df is not None:
logging.info(f"Loaded {symbol_pair} ({interval}) from cache.")
return cached_df
logging.info(f"Fetching {symbol_pair} ({interval}) from Alpha Vantage...")
try:
# Example using TimeSeries daily adjusted - adapt for FX endpoint
# ts = TimeSeries(key=api_key, output_format='pandas')
# data, meta_data = ts.get_daily_adjusted(symbol=from_symbol, outputsize=outputsize) # Adjust function call
# Placeholder for actual FX call structure
fx = ForeignExchange(key=api_key, output_format='pandas')
# Adjust call based on desired interval (e.g., get_currency_exchange_intraday)
data, meta_data = fx.get_currency_exchange_daily(from_symbol=from_symbol, to_symbol=to_symbol, outputsize=outputsize)
# Basic processing: Rename columns for consistency if needed
data.columns = ['Open', 'High', 'Low', 'Close', 'Volume'] # Adjust based on actual AV output
data.index = pd.to_datetime(data.index)
data = data.astype(float)
data.sort_index(inplace=True)
_save_to_cache(data, cache_key, interval)
return data
except Exception as e:
logging.error(f"Alpha Vantage fetch error for {symbol_pair}: {e}")
# Implement retry logic or better error handling
time.sleep(60) # Basic rate limit avoidance
return None
# --- Yahoo Finance Fetching ---
def get_fx_data_yfinance(symbol, period='1y', interval='1d'):
"""Fetches FX data from Yahoo Finance using yfinance with basic caching."""
cache_key = f"YF_{symbol}_{interval}"
cached_df = _load_from_cache(cache_key, interval)
if cached_df is not None:
logging.info(f"Loaded {symbol} ({interval}) from cache.")
return cached_df
logging.info(f"Fetching {symbol} ({interval}) from Yahoo Finance...")
try:
ticker = yf.Ticker(f"{symbol}=X") # Append '=X' for FX pairs
data = ticker.history(period=period, interval=interval)
if data.empty:
logging.warning(f"No data returned for {symbol} from yfinance.")
return None
# Select standard columns
data = data[['Open', 'High', 'Low', 'Close', 'Volume']]
data.index = pd.to_datetime(data.index) # Ensure datetime index
_save_to_cache(data, cache_key, interval)
return data
except Exception as e:
logging.error(f"yfinance fetch error for {symbol}: {e}")
return None
# --- Main Data Function (Used by Agents) ---
def get_fx_data(source='yfinance', **kwargs):
"""Generic function to fetch data based on source."""
if source == 'yfinance':
# Expects 'symbol', 'period', 'interval' in kwargs
return get_fx_data_yfinance(**kwargs)
elif source == 'alphavantage':
# Expects 'api_key', 'from_symbol', 'to_symbol', 'interval', 'outputsize'
return get_fx_data_alphavantage(**kwargs)
else:
logging.error(f"Unsupported data source: {source}")
return None
# Initialize cache directory
# if not os.path.exists(CACHE_DIR):
# os.makedirs(CACHE_DIR)
2. technical_analyzer.py
Purpose: Wraps a chosen technical analysis library (pandas-ta recommended) to provide simple functions for adding indicators to DataFrames.
Functionality: Functions like add_rsi, add_macd, add_bollinger_bands take a DataFrame and parameters, calculate the indicator using the underlying library, and return the DataFrame with new columns.
Libraries: pandas, pandas_ta (or ta)
Python
# technical_analyzer.py (Conceptual Structure using pandas_ta)
import pandas as pd
import pandas_ta as pta
import logging
logging.basicConfig(level=logging.INFO)
def add_rsi(df, period=14):
"""Adds RSI indicator to the DataFrame."""
try:
df.ta.rsi(length=period, append=True)
# Column name might be 'RSI_14' by default
logging.debug(f"Added RSI_{period}")
return df
except Exception as e:
logging.error(f"Error calculating RSI: {e}")
return df # Return original df on error
def add_macd(df, fast=12, slow=26, signal=9):
"""Adds MACD indicator to the DataFrame."""
try:
df.ta.macd(fast=fast, slow=slow, signal=signal, append=True)
# Columns: MACD_12_26_9, MACDh_12_26_9, MACDs_12_26_9
logging.debug(f"Added MACD_{fast}_{slow}_{signal}")
return df
except Exception as e:
logging.error(f"Error calculating MACD: {e}")
return df
def add_bollinger_bands(df, period=20, std_dev=2):
"""Adds Bollinger Bands indicator to the DataFrame."""
try:
df.ta.bbands(length=period, std=std_dev, append=True)
# Columns: BBL_20_2.0, BBM_20_2.0, BBU_20_2.0, BBB_20_2.0, BBP_20_2.0
logging.debug(f"Added BBands_{period}_{std_dev}")
return df
except Exception as e:
logging.error(f"Error calculating Bollinger Bands: {e}")
return df
def add_sma(df, period=20):
"""Adds Simple Moving Average indicator to the DataFrame."""
try:
df.ta.sma(length=period, append=True)
# Column: SMA_20
logging.debug(f"Added SMA_{period}")
return df
except Exception as e:
logging.error(f"Error calculating SMA: {e}")
return df
# Add more functions for other indicators as needed...
def calculate_zscore(df, period=20, column='Close'):
"""Calculates the Z-score relative to a moving average."""
try:
rolling_mean = df[column].rolling(window=period).mean()
rolling_std = df[column].rolling(window=period).std()
df = (df[column] - rolling_mean) / rolling_std
logging.debug(f"Added ZScore_{period}")
return df
except Exception as e:
logging.error(f"Error calculating Z-score: {e}")
return df
def calculate_momentum(df, period=20, column='Close'):
"""Calculates percentage return over a period."""
try:
df[f'Momentum_{period}'] = df[column].pct_change(periods=period)
logging.debug(f"Added Momentum_{period}")
return df
except Exception as e:
logging.error(f"Error calculating Momentum: {e}")
return df
# Example Usage (for testing)
# if __name__ == '__main__':
# # Create a sample DataFrame (replace with actual data loading)
# data = {'Close': [10, 11, 12, 13, 12, 11, 10, 11, 12, 13, 14, 15]}
# sample_df = pd.DataFrame(data)
# sample_df = add_rsi(sample_df)
# sample_df = add_sma(sample_df, period=5)
# print(sample_df)
3. lstm_model_handler.py (If FXSeer LSTM is implemented)
Purpose: Handles loading the pre-trained LSTM model and its associated scaler, preprocessing input data, and running predictions.
Functionality: Separate functions for loading (e.g., using torch.load or tf.keras.models.load_model), scaling input data using the loaded scaler, running the model's predict method, and inverse scaling the output.
Libraries: pytorch or tensorflow, numpy, joblib or pickle (for scaler)
Python
# lstm_model_handler.py (Conceptual Structure - PyTorch Example)
import torch
import numpy as np
import joblib # Or pickle
import logging
logging.basicConfig(level=logging.INFO)
def load_lstm_model(model_path):
"""Loads a pre-trained PyTorch LSTM model."""
try:
# Assuming the model is saved using torch.save(model.state_dict(), path)
# Adjust loading based on how the model was saved (full model vs state_dict)
model = # Define your LSTM model class structure here
model.load_state_dict(torch.load(model_path))
model.eval() # Set model to evaluation mode
logging.info(f"LSTM model loaded from {model_path}")
return model
except Exception as e:
logging.error(f"Error loading LSTM model from {model_path}: {e}")
return None
def load_scaler(scaler_path):
"""Loads a pre-fitted scaler object."""
try:
scaler = joblib.load(scaler_path)
logging.info(f"Scaler loaded from {scaler_path}")
return scaler
except Exception as e:
logging.error(f"Error loading scaler from {scaler_path}: {e}")
return None
def preprocess_for_lstm(data, scaler, lookback):
"""Prepares the most recent data sequence for LSTM prediction."""
try:
# Ensure data is numpy array
data_np = np.array(data).reshape(-1, 1)
# Scale the data
scaled_data = scaler.transform(data_np)
# Get the last 'lookback' sequence
if len(scaled_data) < lookback:
logging.warning("Not enough data for lookback period.")
return None
input_sequence = scaled_data[-lookback:]
# Reshape for LSTM input (e.g., [1, lookback, num_features])
# Adjust shape based on your model's expected input
input_tensor = torch.tensor(input_sequence, dtype=torch.float32).unsqueeze(0)
return input_tensor
except Exception as e:
logging.error(f"Error preprocessing data for LSTM: {e}")
return None
def predict_with_lstm(model, input_tensor):
"""Generates a prediction using the loaded LSTM model."""
if model is None or input_tensor is None:
return None
try:
with torch.no_grad(): # Disable gradient calculation for inference
prediction_scaled = model(input_tensor)
# Extract the prediction value (adjust based on model output shape)
return prediction_scaled.numpy().flatten()
except Exception as e:
logging.error(f"Error during LSTM prediction: {e}")
return None
def inverse_transform_prediction(prediction_scaled, scaler):
"""Inverse transforms the scaled prediction."""
try:
# Scaler expects a 2D array
prediction_reshaped = np.array([[prediction_scaled]])
prediction_original = scaler.inverse_transform(prediction_reshaped)
return prediction_original.flatten()
except Exception as e:
logging.error(f"Error inverse transforming prediction: {e}")
return None
# --- Combined Prediction Function ---
def get_lstm_prediction(model, scaler, recent_data, lookback):
"""Orchestrates preprocessing, prediction, and inverse transform."""
input_tensor = preprocess_for_lstm(recent_data, scaler, lookback)
if input_tensor is None:
return None
prediction_scaled = predict_with_lstm(model, input_tensor)
if prediction_scaled is None:
return None
prediction_original = inverse_transform_prediction(prediction_scaled, scaler)
return prediction_original
B. Prediction Agent Modules
1. fxseer_lstm_agent.py
Structure: Defines FXSeerLSTMAgent class. Assumes an Agno base class or standard methods like run().
__init__: Initializes data source connection parameters, model/scaler paths, lookback period. Loads model and scaler using lstm_model_handler.
run/predict: Orchestrates data fetching (data_fetcher), preparing the latest sequence, calling lstm_model_handler.get_lstm_prediction, and formatting the output dictionary.
Python
# fxseer_lstm_agent.py (Conceptual Structure)
import logging
import datetime
from data_fetcher import get_fx_data
from lstm_model_handler import load_lstm_model, load_scaler, get_lstm_prediction
# Assumes Agno provides a base class or a standard execution method like run()
# from agno_framework import BaseAgent # Hypothetical
logging.basicConfig(level=logging.INFO)
class FXSeerLSTMAgent: # Potentially: FXSeerLSTMAgent(BaseAgent)
def __init__(self, config):
"""Initializes the agent with configuration."""
self.agent_name = "FXSeer LSTM"
self.config = config
self.symbol = config['symbol'] # e.g., 'EURUSD' for AV, 'EURUSD' for YF
self.from_symbol = config.get('from_symbol', 'EUR') # For AV
self.to_symbol = config.get('to_symbol', 'USD') # For AV
self.interval = config['interval']
self.lookback = config['lookback']
self.data_source = config['data_source'] # 'alphavantage' or 'yfinance'
self.api_key = config.get('api_key') # Required for Alpha Vantage
logging.info(f"Initializing {self.agent_name} for {self.symbol}")
# Load model and scaler
self.model = load_lstm_model(config['model_path'])
self.scaler = load_scaler(config['scaler_path'])
if not self.model or not self.scaler:
raise ValueError("Failed to load LSTM model or scaler.")
def run(self):
"""Main execution loop/method for the agent."""
logging.info(f"{self.agent_name} starting run cycle.")
# 1. Fetch Data
fetch_params = {
'source': self.data_source,
'interval': self.interval,
}
if self.data_source == 'yfinance':
fetch_params.update({'symbol': self.symbol, 'period': f"{self.lookback+5}d"}) # Fetch slightly more
elif self.data_source == 'alphavantage':
fetch_params.update({
'api_key': self.api_key,
'from_symbol': self.from_symbol,
'to_symbol': self.to_symbol,
'outputsize': 'compact' # Fetch recent data
})
else:
logging.error("Invalid data source configured.")
return None
df = get_fx_data(**fetch_params)
if df is None or df.empty or len(df) < self.lookback:
logging.warning("Insufficient data for prediction.")
return None
# 2. Prepare Data for Prediction (using 'Close' price)
recent_data = df['Close'].values[-self.lookback:] # Get the last 'lookback' points
# 3. Get Prediction
predicted_price = get_lstm_prediction(
self.model, self.scaler, recent_data, self.lookback
)
if predicted_price is None:
logging.warning("Prediction failed.")
return None
# 4. Format Output
output = {
"agent_name": self.agent_name,
"timestamp": datetime.datetime.utcnow().isoformat() + "Z",
"asset": self.symbol, # Or construct from from/to symbols
"prediction_horizon": f"1{self.interval[-1]}", # e.g., 1D, 1h
"predicted_price": round(float(predicted_price), 5), # Ensure serializable
# "confidence": 0.75 # Placeholder - add if model provides uncertainty
}
logging.info(f"Prediction generated: {output}")
return output # Return prediction for Agno framework
# Example Configuration (load from file e.g., YAML)
# config = {
# 'symbol': 'EURUSD',
# 'from_symbol': 'EUR',
# 'to_symbol': 'USD',
# 'interval': '1d', # Match interval model was trained on
# 'lookback': 20,
# 'data_source': 'yfinance', # or 'alphavantage'
# 'api_key': 'YOUR_AV_KEY_IF_NEEDED',
# 'model_path': 'models/eurusd_lstm.pth', # Or.h5 for TF
# 'scaler_path': 'models/eurusd_scaler.joblib'
# }
# Example Instantiation and Run (if running standalone)
# if __name__ == '__main__':
# agent = FXSeerLSTMAgent(config)
# prediction = agent.run()
# if prediction:
# print(prediction)
2. trendsight_ta_agent.py
Structure: Defines TrendSightTAAgent class.
__init__: Initializes data source, symbol, and specific indicator configurations (periods, thresholds, logic type). Initializes data_fetcher and technical_analyzer.
run/predict: Fetches data, calls technical_analyzer to add required indicators, applies the configured logic (e.g., checks RSI thresholds, MACD crossover), and formats the output dictionary (predicted direction/condition, basis).
Python
# trendsight_ta_agent.py (Conceptual Structure)
import logging
import datetime
import pandas as pd
from data_fetcher import get_fx_data
import technical_analyzer as ta # Use alias
# from agno_framework import BaseAgent # Hypothetical
logging.basicConfig(level=logging.INFO)
class TrendSightTAAgent: # Potentially: TrendSightTAAgent(BaseAgent)
def __init__(self, config):
"""Initializes the agent with configuration."""
self.agent_name = "TrendSight TA"
self.config = config
self.symbol = config['symbol']
self.from_symbol = config.get('from_symbol') # For AV if needed
self.to_symbol = config.get('to_symbol') # For AV if needed
self.interval = config['interval']
self.indicator_config = config['indicator_config'] # e.g., {'name': 'RSI', 'period': 14,...}
self.logic = config['logic'] # e.g., 'RSI_threshold'
self.data_source = config['data_source']
self.api_key = config.get('api_key')
logging.info(f"Initializing {self.agent_name} for {self.symbol} using {self.logic}")
def _apply_logic(self, df):
"""Applies the configured prediction logic."""
latest = df.iloc[-1]
previous = df.iloc[-2] # Needed for crossovers
if self.logic == 'RSI_threshold':
period = self.indicator_config.get('period', 14)
high_thresh = self.indicator_config.get('threshold_high', 70)
low_thresh = self.indicator_config.get('threshold_low', 30)
rsi_col = f'RSI_{period}'
if rsi_col not in latest or pd.isna(latest[rsi_col]): return None, None
rsi_val = latest[rsi_col]
if rsi_val > high_thresh: return 'OVERBOUGHT', f'RSI({period}) > {high_thresh}'
if rsi_val < low_thresh: return 'OVERSOLD', f'RSI({period}) < {low_thresh}'
return 'NEUTRAL', f'{low_thresh} < RSI({period}) < {high_thresh}'
elif self.logic == 'MACD_crossover':
f, s, sig = 12, 26, 9 # Defaults from ta.add_macd
macd_col = f'MACD_{f}_{s}_{sig}'
signal_col = f'MACDs_{f}_{s}_{sig}'
if macd_col not in latest or signal_col not in latest or \
pd.isna(latest[macd_col]) or pd.isna(latest[signal_col]) or \
pd.isna(previous[macd_col]) or pd.isna(previous[signal_col]):
return None, None
# Bullish crossover
if previous[macd_col] < previous[signal_col] and latest[macd_col] > latest[signal_col]:
return 'UP', 'MACD Bullish Crossover'
# Bearish crossover
if previous[macd_col] > previous[signal_col] and latest[macd_col] < latest[signal_col]:
return 'DOWN', 'MACD Bearish Crossover'
return 'NEUTRAL', 'No MACD Crossover'
# Add more logic cases (e.g., BBands_break) here...
else:
logging.warning(f"Unsupported logic type: {self.logic}")
return None, None
def run(self):
"""Main execution loop/method."""
logging.info(f"{self.agent_name} starting run cycle.")
# 1. Fetch Data (fetch enough for indicator calculation)
# Determine required history based on indicator periods
required_periods = 50 # Default, adjust based on max indicator period
if 'period' in self.indicator_config: required_periods = self.indicator_config['period'] * 2
if self.logic == 'MACD_crossover': required_periods = max(required_periods, 35) # MACD needs ~35 periods
fetch_params = {
'source': self.data_source,
'interval': self.interval,
}
if self.data_source == 'yfinance':
fetch_params.update({'symbol': self.symbol, 'period': f"{required_periods+5}d"}) # Fetch more
elif self.data_source == 'alphavantage':
# AV 'compact' is 100 points, 'full' is much more. Use 'compact' usually.
fetch_params.update({
'api_key': self.api_key, 'from_symbol': self.from_symbol, 'to_symbol': self.to_symbol,
'outputsize': 'compact' if required_periods <= 100 else 'full'
})
else: # Handle error
return None
df = get_fx_data(**fetch_params)
if df is None or df.empty or len(df) < required_periods:
logging.warning("Insufficient data for TA calculation.")
return None
# 2. Calculate Indicators
if self.logic == 'RSI_threshold':
df = ta.add_rsi(df, period=self.indicator_config.get('period', 14))
elif self.logic == 'MACD_crossover':
df = ta.add_macd(df) # Use defaults or get from config
# Add elif for other indicators based on self.logic
if df is None: # Check if indicator calculation failed
logging.error("Failed to calculate technical indicators.")
return None
# 3. Apply Logic
prediction, basis = self._apply_logic(df)
if prediction is None:
logging.info("Prediction logic did not yield a result.")
return None # Or return a NEUTRAL state if desired
# 4. Format Output
output = {
"agent_name": self.agent_name,
"timestamp": datetime.datetime.utcnow().isoformat() + "Z",
"asset": self.symbol,
"prediction_horizon": f"Next {self.interval}",
"predicted_direction": prediction, # UP/DOWN/SIDEWAYS/OVERBOUGHT/OVERSOLD
# "signal_strength": "Moderate", # Optional
"basis": basis
}
logging.info(f"Prediction generated: {output}")
return output
# Example Config
# config = {
# 'symbol': 'GBPUSD', 'interval': '1h', 'data_source': 'yfinance',
# 'indicator_config': {'period': 14, 'threshold_high': 70, 'threshold_low': 30},
# 'logic': 'RSI_threshold'
# }
# config = {
# 'symbol': 'USDJPY', 'interval': '4h', 'data_source': 'yfinance',
# 'indicator_config': {}, # MACD uses defaults
# 'logic': 'MACD_crossover'
# }
C. Trading Idea Agent Modules
1. momentum_mover_agent.py
Structure: Defines MomentumMoverAgent class.
__init__: Takes list of symbols, momentum period, N for top/bottom. Initializes data_fetcher.
run/generate_ideas: Fetches data for all symbols, calculates momentum using technical_analyzer or direct pandas, ranks pairs, selects top/bottom N, formats list of output dictionaries.
Python
# momentum_mover_agent.py (Conceptual Structure)
import logging
import datetime
import pandas as pd
from data_fetcher import get_fx_data
import technical_analyzer as ta # For momentum calc if desired
# from agno_framework import BaseAgent # Hypothetical
logging.basicConfig(level=logging.INFO)
class MomentumMoverAgent: # Potentially: MomentumMoverAgent(BaseAgent)
def __init__(self, config):
"""Initializes the agent."""
self.agent_name = "Momentum Mover"
self.config = config
self.symbols = config['symbols'] # List of symbols (e.g.,)
self.from_symbols = config.get('from_symbols') # List if using AV
self.to_symbols = config.get('to_symbols') # List if using AV
self.interval = config['interval'] # Should match momentum period basis (e.g., '1d' for '1M')
self.momentum_period_str = config['momentum_period_str'] # e.g., '1M', '3M', '20d'
self.momentum_period_int = self._parse_period(self.momentum_period_str) # Convert '1M' to days/bars
self.top_n = config['top_n']
self.bottom_n = config['bottom_n']
self.data_source = config['data_source']
self.api_key = config.get('api_key')
logging.info(f"Initializing {self.agent_name} for {len(self.symbols)} symbols, period {self.momentum_period_str}")
def _parse_period(self, period_str):
# Simple parser, assumes daily data if 'M' or 'y' used
num = int(period_str[:-1])
unit = period_str[-1].lower()
if unit == 'd': return num
if unit == 'm': return num * 21 # Approx trading days in month
if unit == 'y': return num * 252 # Approx trading days in year
raise ValueError(f"Invalid period string: {period_str}")
def run(self):
"""Main execution loop."""
logging.info(f"{self.agent_name} starting run cycle.")
momentum_results = {}
# 1. Fetch Data and Calculate Momentum for each symbol
for i, symbol in enumerate(self.symbols):
fetch_params = {
'source': self.data_source, 'interval': self.interval,
}
# Determine required history (momentum period + buffer)
required_history = f"{self.momentum_period_int + 5}d" # Example for daily
if self.data_source == 'yfinance':
fetch_params.update({'symbol': symbol, 'period': required_history})
elif self.data_source == 'alphavantage':
fetch_params.update({
'api_key': self.api_key,
'from_symbol': self.from_symbols[i], 'to_symbol': self.to_symbols[i],
'outputsize': 'compact' # Adjust if period > 100
})
else: continue # Handle error
df = get_fx_data(**fetch_params)
if df is None or df.empty or len(df) < self.momentum_period_int:
logging.warning(f"Insufficient data for {symbol}")
continue
# Calculate momentum using pct_change
# df = ta.calculate_momentum(df, period=self.momentum_period_int) # Or use TA module
df[f'Momentum_{self.momentum_period_int}'] = df['Close'].pct_change(periods=self.momentum_period_int)
if f'Momentum_{self.momentum_period_int}' in df.columns and not pd.isna(df.iloc[-1][f'Momentum_{self.momentum_period_int}']):
momentum_results[symbol] = df.iloc[-1][f'Momentum_{self.momentum_period_int}']
else:
logging.warning(f"Momentum calculation failed for {symbol}")
if not momentum_results:
logging.warning("Momentum calculation failed for all symbols.")
return
# 2. Rank Symbols
sorted_symbols = sorted(momentum_results.items(), key=lambda item: item[1], reverse=True)
# 3. Generate Ideas
trading_ideas =
timestamp = datetime.datetime.utcnow().isoformat() + "Z"
# Top N Buy Ideas
for i in range(min(self.top_n, len(sorted_symbols))):
symbol, momentum = sorted_symbols[i]
if momentum > 0: # Optional: Only generate BUY for positive momentum
ideas = {
"agent_name": self.agent_name, "timestamp": timestamp, "asset": symbol,
"idea_type": "BUY",
"rationale": f"Top {i+1} momentum ({self.momentum_period_str}: {momentum:.2%})",
"confidence": 0.8 # Placeholder
}
trading_ideas.append(ideas)
# Bottom N Sell Ideas
for i in range(min(self.bottom_n, len(sorted_symbols))):
symbol, momentum = sorted_symbols[-(i+1)]
if momentum < 0: # Optional: Only generate SELL for negative momentum
ideas = {
"agent_name": self.agent_name, "timestamp": timestamp, "asset": symbol,
"idea_type": "SELL",
"rationale": f"Bottom {i+1} momentum ({self.momentum_period_str}: {momentum:.2%})",
"confidence": 0.7 # Placeholder
}
trading_ideas.append(ideas)
logging.info(f"Generated {len(trading_ideas)} trading ideas.")
return trading_ideas
# Example Config
# config = {
# 'symbols':,
# 'from_symbols':, # If using AV
# 'to_symbols':, # If using AV
# 'interval': '1d',
# 'momentum_period_str': '1M',
# 'top_n': 1,
# 'bottom_n': 1,
# 'data_source': 'yfinance',
# 'api_key': None
# }
2. mean_reversion_scout_agent.py
Structure: Defines MeanReversionScoutAgent class.
__init__: Takes symbol, lookback period, threshold, method (ZScore/BBands). Initializes data_fetcher, technical_analyzer.
run/generate_ideas: Fetches data, calculates Z-score or BBands using technical_analyzer, checks if price exceeds thresholds, formats output dictionary if signal occurs.
Python
# mean_reversion_scout_agent.py (Conceptual Structure)
import logging
import datetime
import pandas as pd
from data_fetcher import get_fx_data
import technical_analyzer as ta
# from agno_framework import BaseAgent # Hypothetical
logging.basicConfig(level=logging.INFO)
class MeanReversionScoutAgent: # Potentially: MeanReversionScoutAgent(BaseAgent)
def __init__(self, config):
"""Initializes the agent."""
self.agent_name = "Mean Reversion Scout"
self.config = config
self.symbol = config['symbol']
self.from_symbol = config.get('from_symbol') # For AV
self.to_symbol = config.get('to_symbol') # For AV
self.interval = config['interval']
self.lookback_period = config['lookback_period']
self.threshold = config['threshold'] # e.g., 2.0 for Z-score or BBand std dev
self.method = config['method'] # 'ZScore' or 'BollingerBands'
self.data_source = config['data_source']
self.api_key = config.get('api_key')
logging.info(f"Initializing {self.agent_name} for {self.symbol} using {self.method}")
def run(self):
"""Main execution loop."""
logging.info(f"{self.agent_name} starting run cycle.")
# 1. Fetch Data (enough for lookback)
required_periods = self.lookback_period + 5 # Buffer
fetch_params = {
'source': self.data_source, 'interval': self.interval,
}
if self.data_source == 'yfinance':
fetch_params.update({'symbol': self.symbol, 'period': f"{required_periods}d"}) # Adjust period format if needed
elif self.data_source == 'alphavantage':
fetch_params.update({
'api_key': self.api_key, 'from_symbol': self.from_symbol, 'to_symbol': self.to_symbol,
'outputsize': 'compact' # Adjust if period > 100
})
else: return None # Handle error
df = get_fx_data(**fetch_params)
if df is None or df.empty or len(df) < self.lookback_period:
logging.warning("Insufficient data for mean reversion calculation.")
return None
# 2. Calculate Z-Score or Bollinger Bands
idea_type = None
rationale = None
latest_price = df['Close'].iloc[-1]
if self.method == 'ZScore':
df = ta.calculate_zscore(df, period=self.lookback_period)
zscore_col = f'ZScore_{self.lookback_period}'
if zscore_col not in df.columns or pd.isna(df.iloc[-1][zscore_col]):
logging.warning("Z-score calculation failed.")
return None
latest_zscore = df.iloc[-1][zscore_col]
if latest_zscore < -abs(self.threshold):
idea_type = 'BUY'
rationale = f'Z-score < {-abs(self.threshold):.2f} ({latest_zscore:.2f})'
elif latest_zscore > abs(self.threshold):
idea_type = 'SELL'
rationale = f'Z-score > {abs(self.threshold):.2f} ({latest_zscore:.2f})'
elif self.method == 'BollingerBands':
df = ta.add_bollinger_bands(df, period=self.lookback_period, std_dev=self.threshold)
lower_band_col = f'BBL_{self.lookback_period}_{self.threshold}'
upper_band_col = f'BBU_{self.lookback_period}_{self.threshold}'
if lower_band_col not in df.columns or upper_band_col not in df.columns or \
pd.isna(df.iloc[-1][lower_band_col]) or pd.isna(df.iloc[-1][upper_band_col]):
logging.warning("Bollinger Band calculation failed.")
return None
latest_lower_band = df.iloc[-1][lower_band_col]
latest_upper_band = df.iloc[-1][upper_band_col]
if latest_price < latest_lower_band:
idea_type = 'BUY'
rationale = f'Price < Lower BBand ({latest_lower_band:.4f})'
elif latest_price > latest_upper_band:
idea_type = 'SELL'
rationale = f'Price > Upper BBand ({latest_upper_band:.4f})'
else:
logging.error(f"Unsupported method: {self.method}")
return None
# 3. Format Output (only if signal triggered)
if idea_type:
output = {
"agent_name": self.agent_name,
"timestamp": datetime.datetime.utcnow().isoformat() + "Z",
"asset": self.symbol,
"idea_type": idea_type,
"rationale": rationale,
"suggested_entry": round(float(latest_price), 5),
"confidence": 0.7 # Placeholder
}
logging.info(f"Generated trading idea: {output}")
return output
else:
logging.info("No mean reversion signal detected.")
return None # No idea generated
# Example Config
# config = {
# 'symbol': 'USDCAD', 'interval': '1h', 'data_source': 'yfinance',
# 'lookback_period': 20, 'threshold': 2.0, 'method': 'ZScore'
# }
# config = {
# 'symbol': 'AUDUSD', 'interval': '4h', 'data_source': 'yfinance',
# 'lookback_period': 20, 'threshold': 2.0, 'method': 'BollingerBands'
# }
D. Configuration Management
To avoid hardcoding parameters like API keys, symbols, indicator settings, and model paths directly into the agent code, a configuration management approach is essential. This aligns with practices seen in example repositories. Options include:
Environment Variables: Store sensitive information like API keys in environment variables. The os module in Python can access these.
Configuration Files: Use a structured file format like YAML (config.yaml) or INI (config.ini ) or JSON to store all agent parameters. Libraries like PyYAML or Python's built-in configparser can parse these files. The agent's __init__ method would then load its specific configuration section from this file. This approach is generally preferred for managing multiple parameters cleanly.
Example config.yaml structure:
YAML
api_keys:
alpha_vantage: YOUR_AV_KEY_HERE
agents:
fxseer_lstm_eurusd:
agent_class: FXSeerLSTMAgent
symbol: EURUSD
from_symbol: EUR
to_symbol: USD
interval: 1d
lookback: 20
data_source: alphavantage
model_path: models/eurusd_lstm.pth
scaler_path: models/eurusd_scaler.joblib
trendsight_ta_gbpusd:
agent_class: TrendSightTAAgent
symbol: GBPUSD
interval: 1h
data_source: yfinance
indicator_config: {period: 14, threshold_high: 70, threshold_low: 30}
logic: RSI_threshold
momentum_mover:
agent_class: MomentumMoverAgent
symbols:
interval: 1d
momentum_period_str: 1M
top_n: 1
bottom_n: 1
data_source: yfinance
mean_reversion_usdcad:
agent_class: MeanReversionScoutAgent
symbol: USDCAD
interval: 4h
data_source: yfinance
lookback_period: 20
threshold: 2.0
method: ZScore
The Agno framework runner would need to parse this configuration and instantiate the appropriate agents with their respective settings.