Monitor trending momentum of sector ETFs
Looking to build a gemini ai agent based process to monitor trending momentum of sector ETFs to see which ETFs perform better than others. Sector ETF Ticker - Information Technology XLK, Financials XLF, Health Care XLV, Communication Services XLC, Consumer Discretionary XLY, Industrials XLI, Consumer Staples XLP, Energy XLE, Utilities XLU, Materials XLB, Real Estate XLRE
Building a Gemini AI agent-driven pipeline to track and rank sector ETF momentum is a powerful way to automate quantitative market analysis. By combining Python (for fetching price data and calculating metrics) with a Gemini Agent framework (using the google-genai SDK or Agent Development Kit), you can create an autonomous system that ingests market data, computes momentum scores, interprets trend shifts, and delivers structured advisory updates.
Below is a complete blueprint and production-ready architecture for your Gemini-powered sector ETF momentum monitor.
Step 1: Core Python Data & Momentum Engine
This script pulls historical adjusted close prices for your 11 sector ETFs, computes the standard 12-minus-1 month momentum factor (skipping the most recent month to avoid short-term market noise), ranks them, and identifies the leaders.
import pandas as pd
import yfinance as yf
# Define the 11 sector ETF universe
SECTOR_ETFS = {
'Information Technology': 'XLK',
'Financials': 'XLF',
'Health Care': 'XLV',
'Communication Services': 'XLC',
'Consumer Discretionary': 'XLY',
'Industrials': 'XLI',
'Consumer Staples': 'XLP',
'Energy': 'XLE',
'Utilities': 'XLU',
'Materials': 'XLB',
'Real Estate': 'XLRE',
}
def fetch_and_calculate_momentum():
tickers = list(SECTOR_ETFS.values())
# Download historical adjusted close prices (approx. 1.5 years for 12m lookback)
data = yf.download(tickers, period='2y', progress=False)['Close']
# Resample to monthly end prices to compute institutional momentum
monthly_data = data.resample('ME').last()
# 12-1 Month Momentum: Return from 12 months ago to 1 month ago
# (skips the most recent month to avoid short-term reversals)
momentum_12_1 = monthly_data.shift(1) / monthly_data.shift(12) - 1
# Get the latest available momentum scores
latest_scores = momentum_12_1.iloc[-1].dropna()
# Rank sectors from highest momentum to lowest
ranked_sectors = latest_scores.sort_values(ascending=False)
# Map tickers back to readable names
ticker_to_name = {v: k for k, v in SECTOR_ETFS.items()}
results = []
for rank, (ticker, score) in enumerate(ranked_sectors.items(), 1):
results.append({
'Rank': rank,
'Sector': ticker_to_name.get(ticker, ticker),
'Ticker': ticker,
'12M-1M Momentum (%)': round(score * 100, 2),
})
return pd.DataFrame(results)
if __name__ == '__main__':
df_ranked = fetch_and_calculate_momentum()
print(df_ranked.to_string(index=False))
Step 2: Integrating the Gemini Agent Layer
To turn this into a Gemini AI agent process, you wrap the data engine inside a tool or execution loop using the google-genai SDK. The Gemini model acts as the reasoning engine, deciding when to call the momentum tool, analyzing structural market shifts, and generating a synthesized executive briefing.
from google import genai
from google.genai import types
# Initialize the Gemini client
client = genai.Client()
# Define a tool function that the Gemini agent can invoke
def get_sector_momentum_rankings() -> str:
"""Calculates and returns the current momentum rankings for all 11 S&P 500 sector ETFs."""
df = fetch_and_calculate_momentum()
return df.to_json(orient='records')
# Configure the agent with tool access
config = types.GenerateContentConfig(
tools=[get_sector_momentum_rankings],
system_instruction=(
"You are an expert quantitative portfolio manager and AI trading assistant. "
"Your job is to analyze sector ETF momentum data, identify market leadership shifts "
"(e.g., rotation from defensive to cyclical sectors), and provide clear, "
"actionable market commentary based strictly on the computed metrics."
),
temperature=0.2,
)
# Prompt the agent to run the analysis
response = client.models.generate_content(
model='gemini-2.5-flash',
contents=(
'Run the sector momentum scan, identify the top 3 leading sector ETFs,'
' evaluate whether leadership is currently concentrated in growth or'
' defensive sectors, and format a concise weekly briefing.'
),
config=config,
)
print(response.text)
Step 3: Actionable Implementation Roadmap
To scale this into a fully automated production pipeline (e.g., running weekly on Google Cloud Run or Cloud Functions), follow this structured plan:
Database Persistence: Set up a Cloud SQL MySQL instance to store historical daily prices, monthly momentum ranks, and agent run logs using composite primary keys (date, ticker).
Automated Event Triggers: Use Cloud Scheduler and Eventarc to trigger your Python script every Monday pre-market.
Agentic RAG & Synthesis: Have the Gemini agent compare the current week's top 3 sector leaders against the previous month's rankings to automatically detect sector rotation anomalies (e.g., capital flowing out of Utilities XLU into Technology XLK).
Alerting Output: Route the Gemini-generated markdown briefing directly to an internal dashboard or notification channel.
Detail steps to create on gemini cloud platform
Building and deploying your sector ETF momentum agent on Google Cloud Platform (GCP) involves leveraging serverless infrastructure, automated scheduling, and the Gemini API via the google-genai SDK.
Here is the detailed, step-by-step guide to setting up, containerizing, and deploying your agentic momentum monitor on GCP.
Step 1: Set Up Your GCP Project & Enable APIs
First, ensure your Google Cloud environment is properly configured with the necessary services and permissions.
Create or Select a Project:
Open the Google Cloud Console.
Create a new project (e.g., etf-momentum-agent-prod) or select an existing one.
Enable Required APIs: Run the following command in your terminal (or enable them via the Cloud Console API Library):
gcloud services enable \
run.googleapis.com \
artifactregistry.googleapis.com \
sqladmin.googleapis.com \
cloudscheduler.googleapis.com \
secretmanager.googleapis.com \
aiplatform.googleapis.com
Set Up Authentication: Ensure you have the Google Cloud CLI installed and authenticated:
Step 2: Provision Cloud SQL MySQL Database
To store your historical daily prices, momentum ranks, and agent run logs, deploy a managed MySQL database.
NOTE: Would use Cloud SQL instance already set up
Create the Cloud SQL Instance:
Bash
gcloud sql instances create etf-db-instance \
--database-version=MYSQL_8_0 \
--tier=db-f1-micro \
--region=us-central1 \
--root-password=YOUR_SECURE_PASSWORD
Create the Database and User:
Bash
gcloud sql databases create momentum_db --instance=etf-db-instance
Secure Credentials in Secret Manager: Store your database connection string and Gemini API key securely:
Bash
echo -n "mysql+pymysql://root:YOUR_SECURE_PASSWORD@<INSTANCE_IP>/momentum_db" | \
gcloud secrets create db-connection-uri --data-file=-
echo -n "YOUR_GEMINI_API_KEY" | \
gcloud secrets create gemini-api-key --data-file=-
Step 3: Containerize the Agent Application
Package your Python momentum script and Gemini agent into a container image using Docker.
Create a requirements.txt:
pandas
yfinance
pymysql
sqlalchemy
google-genai
requests
Create a Dockerfile:
Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["python", "agent_runner.py"]
Build and Push to Artifact Registry: Create an Artifact Registry repository and push your container image:
Bash
gcloud artifacts repositories create agent-repo \
--repository-format=docker \
--location=us-central1
gcloud auth configure-docker us-central1-docker.pkg.dev
docker build -t us-central1-docker.pkg.dev/YOUR_PROJECT_ID/agent-repo/etf-momentum-agent:v1 .
docker push us-central1-docker.pkg.dev/YOUR_PROJECT_ID/agent-repo/etf-momentum-agent:v1
Step 4: Deploy as a Serverless Job on Cloud Run
Cloud Run Jobs allow you to run script-based batch workloads that execute to completion (ideal for weekly or daily market scans).
Deploy the Cloud Run Job:
Bash
gcloud run jobs create etf-momentum-job \
--image=us-central1-docker.pkg.dev/YOUR_PROJECT_ID/agent-repo/etf-momentum-agent:v1 \
--region=us-central1 \
--set-secrets=DATABASE_URL=db-connection-uri:latest,GEMINI_API_KEY=gemini-api-key:latest \
--max-retries=3 \
--task-timeout=5m
Step 5: Automate with Cloud Scheduler
Set up a cron trigger so the agent automatically scans the market and stores results before the market opens every Monday.
Create a Scheduler Trigger:
Bash
gcloud scheduler jobs create http weekly-momentum-trigger \
--location=us-central1 \
--schedule="0 8 * * 1" \
--uri="https://us-central1-run.googleapis.com/apis/run.googleapis.com/v1/namespaces/YOUR_PROJECT_ID/jobs/etf-momentum-job:run" \
--oauth-service-account-email=YOUR_PROJECT_NUMBER-compute@developer.gserviceaccount.com
(Note: This schedules the job to run every Monday at 8:00 AM).
What would be the directory structure
To support a clean, modular, and production-ready Python application that you can easily containerize and deploy to Google Cloud Run, here is the recommended directory structure for your Sector ETF Momentum Agent.
This structure cleanly separates your data ingestion logic, database models, Gemini agent configuration, configuration files, and deployment assets.
Recommended Directory Structure
Key File Descriptions
main.py: The application entrypoint. When triggered by Cloud Run or Cloud Scheduler, it executes the pipeline: fetches the latest momentum rankings via momentum.py, logs the metrics into Cloud SQL via database/, and invokes the Gemini agent in engine/agent.py to produce the executive briefing.
engine/momentum.py: Contains the yfinance download logic, monthly resampling, and 12-minus-1 month return calculations for your 11 sector ETFs (XLK, XLF, XLV, etc.).
engine/agent.py: Configures the google-genai client, wraps your momentum function as a Gemini tool, and manages system instructions for the model.
database/models.py: Defines your SQLAlchemy ORM models with composite primary keys (date, ticker) to track historical momentum scores safely over time.
.dockerignore: Ensures local environment files, caches (__pycache__), and git folders are ignored when building your Docker image.
List out all the python code for each .py file
Here is the complete, production-ready Python code for each file in your directory structure, integrating data ingestion, SQLAlchemy database persistence, and the Gemini AI agent loop.
1. config/settings.py
Handles environment configuration and secret loading.
Python
import os
from dotenv import load_dotenv
load_dotenv()
DATABASE_URL = os.getenv("DATABASE_URL")
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
GEMINI_MODEL = os.getenv("GEMINI_MODEL", "gemini-2.5-flash")
2. database/connection.py
Sets up the SQLAlchemy database engine, session factory, and base model class.
Python
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker
from config.settings import DATABASE_URL
if not DATABASE_URL:
raise ValueError("DATABASE_URL environment variable is not set.")
engine = create_engine(DATABASE_URL, echo=False, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
3. database/models.py
Defines MySQL database tables using SQLAlchemy ORM models.
Python
from datetime import datetime
from sqlalchemy import Column, Date, DateTime, Float, Integer, String, Text
from database.connection import Base
class ETFPriceHistory(Base):
__tablename__ = "etf_price_history"
date = Column(Date, primary_key=True)
ticker = Column(String(10), primary_key=True)
adjusted_close = Column(Float, nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
class MomentumRanking(Base):
__tablename__ = "momentum_rankings"
date = Column(Date, primary_key=True)
ticker = Column(String(10), primary_key=True)
sector_name = Column(String(50), nullable=False)
momentum_score = Column(Float, nullable=False)
rank = Column(Integer, nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
class AgentExecutionLog(Base):
__tablename__ = "agent_execution_logs"
id = Column(Integer, primary_key=True, autoincrement=True)
execution_date = Column(DateTime, default=datetime.utcnow)
status = Column(String(20), nullable=False)
summary_output = Column(Text, nullable=True)
error_message = Column(Text, nullable=True)
4. engine/momentum.py
Downloads historical ETF prices via yfinance and computes institutional 12-minus-1 month momentum scores.
Python
import pandas as pd
import yfinance as yf
SECTOR_ETFS = {
'Information Technology': 'XLK',
'Financials': 'XLF',
'Health Care': 'XLV',
'Communication Services': 'XLC',
'Consumer Discretionary': 'XLY',
'Industrials': 'XLI',
'Consumer Staples': 'XLP',
'Energy': 'XLE',
'Utilities': 'XLU',
'Materials': 'XLB',
'Real Estate': 'XLRE',
}
def fetch_and_calculate_momentum() -> pd.DataFrame:
tickers = list(SECTOR_ETFS.values())
# Download historical adjusted close prices
data = yf.download(tickers, period='2y', progress=False)['Close']
# Resample to monthly end prices
monthly_data = data.resample('ME').last()
# 12-1 Month Momentum: Return from 12 months ago to 1 month ago
momentum_12_1 = monthly_data.shift(1) / monthly_data.shift(12) - 1
latest_scores = momentum_12_1.iloc[-1].dropna()
ranked_sectors = latest_scores.sort_values(ascending=False)
ticker_to_name = {v: k for k, v in SECTOR_ETFS.items()}
results = []
for rank, (ticker, score) in enumerate(ranked_sectors.items(), 1):
results.append({
'Rank': rank,
'Sector': ticker_to_name.get(ticker, ticker),
'Ticker': ticker,
'12M-1M Momentum (%)': round(score * 100, 2),
'Raw Score': float(score),
})
return pd.DataFrame(results)
5. engine/agent.py
Initializes the google-genai client, exposes the momentum function as a tool, and configures the Gemini model.
Python
from google import genai
from google.genai import types
from config.settings import GEMINI_API_KEY, GEMINI_MODEL
from engine.momentum import fetch_and_calculate_momentum
client = genai.Client(api_key=GEMINI_API_KEY) if GEMINI_API_KEY else genai.Client()
def get_sector_momentum_rankings() -> str:
"""Calculates and returns the current momentum rankings for all 11 S&P 500 sector ETFs."""
df = fetch_and_calculate_momentum()
return df.to_json(orient='records')
def run_momentum_agent(custom_prompt: str = None) -> str:
config = types.GenerateContentConfig(
tools=[get_sector_momentum_rankings],
system_instruction=(
"You are an expert quantitative portfolio manager and AI trading"
" assistant. Your job is to analyze sector ETF momentum data, identify"
" market leadership shifts (e.g., rotation from defensive to"
" cyclical sectors), and provide clear, actionable market commentary"
" based strictly on the computed metrics."
),
temperature=0.2,
)
prompt = custom_prompt or (
'Run the sector momentum scan, identify the top 3 leading sector ETFs,'
' evaluate whether leadership is currently concentrated in growth or'
' defensive sectors, and format a concise weekly briefing.'
)
response = client.models.generate_content(
model=GEMINI_MODEL,
contents=prompt,
config=config,
)
return response.text
6. main.py
The orchestration entrypoint that runs the pipeline, writes metrics to Cloud SQL, and triggers the Gemini agent.
Python
from datetime import date
import logging
from database.connection import Base, SessionLocal, engine
from database.models import AgentExecutionLog, MomentumRanking
from engine.agent import run_momentum_agent
from engine.momentum import fetch_and_calculate_momentum
logging.basicConfig(
level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def init_db():
logger.info('Initializing database tables...')
Base.metadata.create_all(bind=engine)
def main():
logger.info('Starting Sector ETF Momentum Agent pipeline...')
init_db()
db = SessionLocal()
try:
logger.info('Fetching market data and calculating momentum...')
df_rankings = fetch_and_calculate_momentum()
today = date.today()
logger.info('Persisting momentum rankings to database...')
for _, row in df_rankings.iterrows():
ranking_record = MomentumRanking(
date=today,
ticker=row['Ticker'],
sector_name=row['Sector'],
momentum_score=row['Raw Score'],
rank=int(row['Rank']),
)
db.merge(ranking_record)
db.commit()
logger.info('Invoking Gemini AI Agent for market commentary...')
agent_output = run_momentum_agent()
logger.info('--- GEMINI AGENT EXECUTIVE BRIEFING ---')
print(agent_output)
logger.info('---------------------------------------')
log_record = AgentExecutionLog(status='SUCCESS', summary_output=agent_output)
db.add(log_record)
db.commit()
logger.info('Pipeline completed successfully.')
except Exception as e:
logger.error(f'Pipeline execution failed: {e}', exc_info=True)
db.rollback()
try:
error_log = AgentExecutionLog(status='FAILED', error_message=str(e))
db.add(error_log)
db.commit()
except Exception as db_err:
logger.error(f'Failed to write error log to DB: {db_err}')
raise e
finally:
db.close()
if __name__ == '__main__':
main()
No comments:
Post a Comment