How to Track Crypto ETF Flows with Ollama: Monitor Institutional Money Movement Like a Pro

# Install Ollama
curl -fsSL https://ollama.ai/install.sh | sh

# Pull the recommended model for financial analysis
ollama pull llama2:13b

# Verify installation
ollama list

# etf_flow_collector.py
import requests
import pandas as pd
from datetime import datetime, timedelta
import json

class CryptoETFFlowTracker:
    def __init__(self):
        self.etf_symbols = ['IBIT', 'GBTC', 'BITB', 'ARKB', 'FBTC']
        self.api_endpoints = {
            'flow_data': 'https://api.etfdb.com/v1/flows',
            'price_data': 'https://api.marketdata.app/v1/stocks/quotes'
        }
    
    def fetch_daily_flows(self, symbol, date):
        """Fetch daily flow data for specific ETF"""
        params = {
            'symbol': symbol,
            'date': date.strftime('%Y-%m-%d'),
            'format': 'json'
        }
        
        response = requests.get(self.api_endpoints['flow_data'], params=params)
        
        if response.status_code == 200:
            return response.json()
        else:
            print(f"Error fetching data for {symbol}: {response.status_code}")
            return None
    
    def calculate_flow_metrics(self, flow_data):
        """Calculate key flow metrics"""
        if not flow_data:
            return None
            
        return {
            'net_flow': flow_data.get('inflow', 0) - flow_data.get('outflow', 0),
            'flow_percentage': (flow_data.get('net_flow', 0) / flow_data.get('aum', 1)) * 100,
            'momentum_score': self.calculate_momentum(flow_data)
        }
    
    def calculate_momentum(self, flow_data):
        """Calculate momentum score based on flow patterns"""
        # Custom momentum calculation
        recent_flows = flow_data.get('recent_flows', [])
        if len(recent_flows) < 5:
            return 0
            
        positive_days = sum(1 for flow in recent_flows if flow > 0)
        return (positive_days / len(recent_flows)) * 100

# Initialize tracker
tracker = CryptoETFFlowTracker()

# ollama_analyzer.py
import subprocess
import json
from datetime import datetime

class OllamaETFAnalyzer:
    def __init__(self, model="llama2:13b"):
        self.model = model
        
    def analyze_flow_patterns(self, etf_data):
        """Use Ollama to analyze ETF flow patterns"""
        
        # Prepare prompt for Ollama
        prompt = f"""
        Analyze these crypto ETF flow patterns and identify key trends:
        
        ETF Data: {json.dumps(etf_data, indent=2)}
        
        Please provide:
        1. Overall market sentiment (bullish/bearish/neutral)
        2. Which ETFs show strongest inflows
        3. Risk factors based on flow patterns
        4. Recommended actions for retail investors
        
        Format your response as JSON with clear categories.
        """
        
        # Call Ollama
        result = subprocess.run([
            'ollama', 'run', self.model, prompt
        ], capture_output=True, text=True)
        
        if result.returncode == 0:
            return self.parse_ollama_response(result.stdout)
        else:
            print(f"Ollama error: {result.stderr}")
            return None
    
    def parse_ollama_response(self, response):
        """Parse and structure Ollama's analysis"""
        try:
            # Extract JSON from response
            start_idx = response.find('{')
            end_idx = response.rfind('}') + 1
            json_str = response[start_idx:end_idx]
            
            return json.loads(json_str)
        except:
            # Fallback to text parsing
            return {'raw_analysis': response}

# Initialize analyzer
analyzer = OllamaETFAnalyzer()

# dashboard.py
import streamlit as st
import plotly.graph_objects as go
import plotly.express as px
from datetime import datetime, timedelta
import pandas as pd

def create_flow_dashboard():
    """Create interactive ETF flow dashboard"""
    
    st.title("🚀 Crypto ETF Flow Tracker")
    st.markdown("Monitor institutional money movement in real-time")
    
    # Sidebar controls
    st.sidebar.header("Dashboard Controls")
    selected_etfs = st.sidebar.multiselect(
        "Select ETFs to Track",
        ['IBIT', 'GBTC', 'BITB', 'ARKB', 'FBTC'],
        default=['IBIT', 'GBTC']
    )
    
    time_range = st.sidebar.selectbox(
        "Time Range",
        ['7 days', '30 days', '90 days', '1 year']
    )
    
    # Main dashboard
    col1, col2, col3 = st.columns(3)
    
    with col1:
        st.metric("Total Net Inflows", "$2.1B", "+15.2%")
    
    with col2:
        st.metric("Largest Single Day Inflow", "$890M", "IBIT")
    
    with col3:
        st.metric("Market Sentiment", "Bullish", "+2 points")
    
    # Flow chart
    st.subheader("📊 Daily Flow Patterns")
    
    # Sample data for demonstration
    dates = pd.date_range(start=datetime.now() - timedelta(days=30), end=datetime.now())
    flow_data = pd.DataFrame({
        'Date': dates,
        'IBIT': [random.randint(-100, 300) for _ in range(len(dates))],
        'GBTC': [random.randint(-200, 100) for _ in range(len(dates))]
    })
    
    fig = px.line(flow_data, x='Date', y=selected_etfs, 
                  title="ETF Flow Trends (Millions USD)")
    st.plotly_chart(fig, use_container_width=True)
    
    # Ollama insights
    st.subheader("🤖 AI Analysis")
    
    if st.button("Generate Ollama Insights"):
        with st.spinner("Analyzing flow patterns with Ollama..."):
            # Simulate Ollama analysis
            insights = {
                "sentiment": "Bullish",
                "key_trends": [
                    "IBIT showing consistent inflows for 7 consecutive days",
                    "GBTC outflows stabilizing after initial ETF conversion period",
                    "Institutional accumulation pattern detected"
                ],
                "recommendations": [
                    "Monitor IBIT for continued momentum",
                    "Watch for GBTC reversal signals",
                    "Consider position sizing based on flow velocity"
                ]
            }
            
            st.json(insights)

if __name__ == "__main__":
    create_flow_dashboard()

def analyze_event_correlations(flow_data, market_events):
    """Analyze how flows respond to market events"""
    
    correlations = {}
    
    for event in market_events:
        event_date = event['date']
        event_impact = calculate_flow_impact(flow_data, event_date)
        
        correlations[event['name']] = {
            'flow_change': event_impact['flow_change'],
            'correlation_strength': event_impact['correlation'],
            'duration': event_impact['impact_duration']
        }
    
    return correlations

# Example usage
market_events = [
    {'name': 'Bitcoin ETF Approval', 'date': '2024-01-10'},
    {'name': 'Fed Rate Decision', 'date': '2024-03-20'},
    {'name': 'Bitcoin Halving', 'date': '2024-04-19'}
]

correlations = analyze_event_correlations(flow_data, market_events)

# alert_system.py
import smtplib
from email.mime.text import MimeText
from email.mime.multipart import MimeMultipart

class ETFFlowAlertSystem:
    def __init__(self, email_config):
        self.email_config = email_config
        self.alert_thresholds = {
            'large_inflow': 100_000_000,  # $100M
            'large_outflow': -50_000_000,  # -$50M
            'momentum_shift': 0.15  # 15% change in momentum
        }
    
    def check_alert_conditions(self, current_flows):
        """Check if any alert conditions are met"""
        alerts = []
        
        for etf, data in current_flows.items():
            # Check large flows
            if data['net_flow'] > self.alert_thresholds['large_inflow']:
                alerts.append({
                    'type': 'large_inflow',
                    'etf': etf,
                    'amount': data['net_flow'],
                    'message': f"🚨 Large inflow detected: {etf} received ${data['net_flow']:,.0f}"
                })
            
            elif data['net_flow'] < self.alert_thresholds['large_outflow']:
                alerts.append({
                    'type': 'large_outflow',
                    'etf': etf,
                    'amount': data['net_flow'],
                    'message': f"⚠️ Large outflow detected: {etf} lost ${abs(data['net_flow']):,.0f}"
                })
        
        return alerts
    
    def send_alert(self, alert):
        """Send email alert for significant flow events"""
        
        msg = MimeMultipart()
        msg['From'] = self.email_config['sender']
        msg['To'] = self.email_config['recipient']
        msg['Subject'] = f"Crypto ETF Alert: {alert['type']}"
        
        body = f"""
        {alert['message']}
        
        ETF: {alert['etf']}
        Flow Amount: ${alert['amount']:,.0f}
        Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
        
        Check your dashboard for detailed analysis.
        """
        
        msg.attach(MimeText(body, 'plain'))
        
        try:
            server = smtplib.SMTP(self.email_config['smtp_server'], 587)
            server.starttls()
            server.login(self.email_config['username'], self.email_config['password'])
            text = msg.as_string()
            server.sendmail(self.email_config['sender'], self.email_config['recipient'], text)
            server.quit()
            print(f"Alert sent successfully for {alert['type']}")
        except Exception as e:
            print(f"Failed to send alert: {e}")

# Example usage
email_config = {
    'sender': 'your-email@gmail.com',
    'recipient': 'alerts@yourdomain.com',
    'username': 'your-email@gmail.com',
    'password': 'your-app-password',
    'smtp_server': 'smtp.gmail.com'
}

alert_system = ETFFlowAlertSystem(email_config)

def clean_flow_data(raw_data):
    """Clean and validate ETF flow data"""
    
    cleaned_data = {}
    
    for etf, data in raw_data.items():
        # Remove outliers
        if abs(data['net_flow']) > 1_000_000_000:  # $1B threshold
            print(f"Warning: Unusual flow detected for {etf}: ${data['net_flow']:,.0f}")
            
        # Validate data completeness
        required_fields = ['net_flow', 'inflow', 'outflow', 'aum']
        if all(field in data for field in required_fields):
            cleaned_data[etf] = data
        else:
            print(f"Incomplete data for {etf}, skipping...")
    
    return cleaned_data

import time
from functools import wraps

def rate_limit(calls_per_minute=60):
    """Decorator to rate limit API calls"""
    min_interval = 60.0 / calls_per_minute
    last_called = [0.0]
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            left_to_wait = min_interval - elapsed
            if left_to_wait > 0:
                time.sleep(left_to_wait)
            ret = func(*args, **kwargs)
            last_called[0] = time.time()
            return ret
        return wrapper
    return decorator

@rate_limit(calls_per_minute=30)
def fetch_etf_data(symbol):
    """Rate-limited data fetching"""
    return tracker.fetch_daily_flows(symbol, datetime.now())