Paul Atkins SEC Policy Tracker: Build AI-Powered Crypto Regulation Analysis Tool with Ollama

# macOS
curl -fsSL https://ollama.com/install.sh | sh

# Linux
curl -fsSL https://ollama.com/install.sh | sh

# Windows (PowerShell)
Invoke-WebRequest -Uri https://ollama.com/install.sh -OutFile install.sh; bash install.sh

# Pull the models
ollama pull deepseek-r1:7b
ollama pull llama3.3:70b
ollama pull qwen2.5:32b

# Verify installation
ollama list

import requests
from bs4 import BeautifulSoup
import json
import datetime
import ollama
from typing import Dict, List, Optional

class SECPolicyTracker:
    def __init__(self, models=['deepseek-r1:7b', 'llama3.3:70b']):
        self.models = models
        self.sec_base_url = 'https://www.sec.gov'
        self.policy_keywords = [
            'digital assets', 'cryptocurrency', 'crypto', 'blockchain',
            'DeFi', 'stablecoin', 'virtual currency', 'Paul Atkins'
        ]
        
    def fetch_sec_announcements(self, days_back: int = 7) -> List[Dict]:
        """Fetch recent SEC announcements and press releases"""
        announcements = []
        
        # Fetch press releases
        press_url = f"{self.sec_base_url}/news/press-releases"
        response = requests.get(press_url)
        
        if response.status_code == 200:
            soup = BeautifulSoup(response.content, 'html.parser')
            
            # Extract announcement links and titles
            for item in soup.find_all('div', class_='views-row'):
                title_elem = item.find('h3')
                if title_elem and title_elem.find('a'):
                    title = title_elem.find('a').text.strip()
                    link = title_elem.find('a')['href']
                    
                    # Check if crypto-related
                    if any(keyword.lower() in title.lower() for keyword in self.policy_keywords):
                        announcements.append({
                            'title': title,
                            'url': f"{self.sec_base_url}{link}",
                            'date': datetime.datetime.now().isoformat(),
                            'type': 'press_release'
                        })
        
        return announcements
    
    def extract_policy_content(self, url: str) -> str:
        """Extract full content from SEC announcement"""
        try:
            response = requests.get(url)
            if response.status_code == 200:
                soup = BeautifulSoup(response.content, 'html.parser')
                
                # Remove navigation and footer elements
                for element in soup(['nav', 'footer', 'script', 'style']):
                    element.decompose()
                
                # Extract main content
                content_div = soup.find('div', class_='field-content') or soup.find('main')
                if content_div:
                    return content_div.get_text(strip=True)
                
                return soup.get_text(strip=True)
        except Exception as e:
            print(f"Error extracting content from {url}: {e}")
            return ""
    
    def analyze_policy_changes(self, content: str, model: str = 'deepseek-r1:7b') -> Dict:
        """Analyze policy content using Ollama"""
        analysis_prompt = f"""
        Analyze this SEC announcement for crypto and digital asset policy changes:
        
        {content}
        
        Provide analysis in this JSON format:
        {{
            "policy_changes": ["list of specific policy changes"],
            "crypto_impact": "high/medium/low",
            "affected_sectors": ["list of affected crypto sectors"],
            "compliance_requirements": ["list of new compliance requirements"],
            "timeline": "implementation timeline if mentioned",
            "key_quotes": ["important quotes from the announcement"],
            "atkins_stance": "Paul Atkins' position on this issue if mentioned"
        }}
        """
        
        try:
            response = ollama.generate(
                model=model,
                prompt=analysis_prompt,
                stream=False
            )
            
            # Parse JSON response
            analysis = json.loads(response['response'])
            return analysis
            
        except Exception as e:
            print(f"Error analyzing with {model}: {e}")
            return {}
    
    def assess_business_impact(self, policy_analysis: Dict, business_type: str) -> Dict:
        """Assess impact on specific crypto business models"""
        impact_prompt = f"""
        Based on this policy analysis, assess the impact on {business_type} businesses:
        
        Policy Analysis: {json.dumps(policy_analysis, indent=2)}
        
        Provide assessment in JSON format:
        {{
            "risk_level": "high/medium/low",
            "opportunities": ["list of business opportunities"],
            "compliance_actions": ["immediate actions needed"],
            "cost_impact": "estimated compliance costs",
            "timeline_to_comply": "time needed to implement changes"
        }}
        """
        
        try:
            response = ollama.generate(
                model='llama3.3:70b',
                prompt=impact_prompt,
                stream=False
            )
            
            return json.loads(response['response'])
            
        except Exception as e:
            print(f"Error assessing business impact: {e}")
            return {}

class PolicyAnalyzer:
    def __init__(self, tracker: SECPolicyTracker):
        self.tracker = tracker
    
    def analyze_digital_asset_classification(self, content: str) -> Dict:
        """Analyze changes to digital asset classification rules"""
        prompt = f"""
        Analyze this SEC content for digital asset classification changes:
        
        {content}
        
        Focus on:
        1. Changes to security token definitions
        2. Utility token classification updates
        3. Safe harbor provisions
        4. Registration exemptions
        5. Paul Atkins' classification framework
        
        Return detailed JSON analysis.
        """
        
        return self.tracker.analyze_policy_changes(content, 'deepseek-r1:7b')
    
    def analyze_defi_regulations(self, content: str) -> Dict:
        """Analyze DeFi-specific regulatory updates"""
        prompt = f"""
        Analyze this SEC content for DeFi regulation changes:
        
        {content}
        
        Focus on:
        1. Smart contract compliance requirements
        2. Decentralized exchange regulations
        3. Yield farming and staking rules
        4. Governance token classification
        5. Protocol liability frameworks
        
        Return detailed JSON analysis.
        """
        
        return self.tracker.analyze_policy_changes(content, 'qwen2.5:32b')
    
    def analyze_institutional_crypto(self, content: str) -> Dict:
        """Analyze institutional crypto investment rules"""
        prompt = f"""
        Analyze this SEC content for institutional crypto changes:
        
        {content}
        
        Focus on:
        1. Custody requirements updates
        2. Investment advisor regulations
        3. Fund management rules
        4. Reporting requirements
        5. Risk management frameworks
        
        Return detailed JSON analysis.
        """
        
        return self.tracker.analyze_policy_changes(content, 'llama3.3:70b')

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import json

class PolicyAlertSystem:
    def __init__(self, smtp_config: Dict):
        self.smtp_config = smtp_config
        self.alert_thresholds = {
            'high_impact': ['high'],
            'medium_impact': ['medium', 'high'],
            'all_changes': ['low', 'medium', 'high']
        }
    
    def should_alert(self, analysis: Dict, threshold: str = 'medium_impact') -> bool:
        """Determine if analysis meets alert threshold"""
        impact_level = analysis.get('crypto_impact', 'low')
        return impact_level in self.alert_thresholds.get(threshold, [])
    
    def format_alert_message(self, analysis: Dict, announcement: Dict) -> str:
        """Format policy analysis into readable alert"""
        message = f"""
        🚨 SEC Policy Alert: {announcement['title']}
        
        📊 Impact Level: {analysis.get('crypto_impact', 'Unknown').upper()}
        
        🏛️ Policy Changes:
        {chr(10).join(f"• {change}" for change in analysis.get('policy_changes', []))}
        
        🎯 Affected Sectors:
        {chr(10).join(f"• {sector}" for sector in analysis.get('affected_sectors', []))}
        
        ⚖️ Compliance Requirements:
        {chr(10).join(f"• {req}" for req in analysis.get('compliance_requirements', []))}
        
        ⏰ Timeline: {analysis.get('timeline', 'Not specified')}
        
        📝 Paul Atkins' Stance: {analysis.get('atkins_stance', 'Not mentioned')}
        
        🔗 Source: {announcement['url']}
        
        ---
        Generated by Paul Atkins SEC Policy Tracker
        """
        
        return message
    
    def send_alert(self, message: str, recipients: List[str]):
        """Send email alert"""
        try:
            msg = MIMEMultipart()
            msg['From'] = self.smtp_config['from_email']
            msg['To'] = ', '.join(recipients)
            msg['Subject'] = "SEC Policy Alert - Crypto Regulation Update"
            
            msg.attach(MIMEText(message, 'plain'))
            
            server = smtplib.SMTP(self.smtp_config['server'], self.smtp_config['port'])
            server.starttls()
            server.login(self.smtp_config['username'], self.smtp_config['password'])
            
            server.send_message(msg)
            server.quit()
            
            print(f"Alert sent to {len(recipients)} recipients")
            
        except Exception as e:
            print(f"Error sending alert: {e}")

#!/usr/bin/env python3
import time
import json
from datetime import datetime

def main():
    # Initialize tracker
    tracker = SECPolicyTracker()
    analyzer = PolicyAnalyzer(tracker)
    
    # Configure alerts
    smtp_config = {
        'server': 'smtp.gmail.com',
        'port': 587,
        'username': 'your-email@gmail.com',
        'password': 'your-app-password',
        'from_email': 'your-email@gmail.com'
    }
    
    alert_system = PolicyAlertSystem(smtp_config)
    recipients = ['legal@yourcompany.com', 'compliance@yourcompany.com']
    
    print("🚀 Starting Paul Atkins SEC Policy Tracker...")
    
    while True:
        try:
            # Fetch new announcements
            announcements = tracker.fetch_sec_announcements(days_back=1)
            
            for announcement in announcements:
                print(f"📄 Processing: {announcement['title']}")
                
                # Extract content
                content = tracker.extract_policy_content(announcement['url'])
                
                if content:
                    # Analyze policy changes
                    analysis = tracker.analyze_policy_changes(content)
                    
                    if analysis:
                        # Specialized analysis based on content type
                        if 'digital asset' in announcement['title'].lower():
                            detailed_analysis = analyzer.analyze_digital_asset_classification(content)
                        elif 'defi' in announcement['title'].lower():
                            detailed_analysis = analyzer.analyze_defi_regulations(content)
                        else:
                            detailed_analysis = analysis
                        
                        # Check if alert needed
                        if alert_system.should_alert(detailed_analysis, 'medium_impact'):
                            alert_message = alert_system.format_alert_message(
                                detailed_analysis, announcement
                            )
                            alert_system.send_alert(alert_message, recipients)
                        
                        # Log analysis
                        log_entry = {
                            'timestamp': datetime.now().isoformat(),
                            'announcement': announcement,
                            'analysis': detailed_analysis
                        }
                        
                        with open('policy_analysis_log.json', 'a') as f:
                            f.write(json.dumps(log_entry) + '\n')
            
            # Sleep for 1 hour before next check
            time.sleep(3600)
            
        except KeyboardInterrupt:
            print("\n🛑 Stopping tracker...")
            break
        except Exception as e:
            print(f"❌ Error in main loop: {e}")
            time.sleep(300)  # Wait 5 minutes before retry

if __name__ == "__main__":
    main()

# Configure Ollama for optimal performance
def optimize_ollama_performance():
    """Configure Ollama for SEC policy analysis workloads"""
    
    # Enable GPU acceleration
    import subprocess
    
    # Check GPU availability
    result = subprocess.run(['nvidia-smi'], capture_output=True, text=True)
    
    if result.returncode == 0:
        print("✅ NVIDIA GPU detected - enabling CUDA acceleration")
        # GPU-optimized model configuration
        model_config = {
            'num_gpu': 1,
            'num_ctx': 8192,  # Longer context for policy documents
            'temperature': 0.1,  # Lower temperature for factual analysis
            'top_k': 10,
            'top_p': 0.9
        }
    else:
        print("⚠️  No GPU detected - using CPU optimization")
        model_config = {
            'num_gpu': 0,
            'num_ctx': 4096,
            'temperature': 0.1,
            'num_thread': 8
        }
    
    return model_config

import streamlit as st
import plotly.graph_objects as go
import pandas as pd
from datetime import datetime, timedelta

def create_dashboard():
    """Create Streamlit dashboard for policy tracking"""
    
    st.set_page_config(
        page_title="Paul Atkins SEC Policy Tracker",
        page_icon="🏛️",
        layout="wide"
    )
    
    st.title("🏛️ Paul Atkins SEC Policy Tracker")
    st.markdown("Monitor crypto-friendly regulatory changes from the new SEC Chair")
    
    # Load recent analyses
    try:
        with open('policy_analysis_log.json', 'r') as f:
            analyses = [json.loads(line) for line in f.readlines()[-50:]]
    except FileNotFoundError:
        analyses = []
    
    if analyses:
        # Create impact distribution chart
        impact_data = {'high': 0, 'medium': 0, 'low': 0}
        for analysis in analyses:
            impact = analysis.get('analysis', {}).get('crypto_impact', 'low')
            impact_data[impact] += 1
        
        fig = go.Figure(data=[
            go.Bar(
                x=list(impact_data.keys()),
                y=list(impact_data.values()),
                marker_color=['red', 'yellow', 'green']
            )
        ])
        
        fig.update_layout(
            title="Policy Impact Distribution (Last 50 Analyses)",
            xaxis_title="Impact Level",
            yaxis_title="Number of Policies"
        )
        
        st.plotly_chart(fig, use_container_width=True)
        
        # Recent policy changes
        st.subheader("📋 Recent Policy Changes")
        
        for analysis in analyses[-10:]:
            announcement = analysis.get('announcement', {})
            policy_analysis = analysis.get('analysis', {})
            
            with st.expander(f"📄 {announcement.get('title', 'Unknown')}"):
                st.write(f"**Impact Level:** {policy_analysis.get('crypto_impact', 'Unknown')}")
                st.write(f"**Date:** {analysis.get('timestamp', 'Unknown')}")
                
                if policy_analysis.get('policy_changes'):
                    st.write("**Policy Changes:**")
                    for change in policy_analysis.get('policy_changes', []):
                        st.write(f"• {change}")
                
                if policy_analysis.get('compliance_requirements'):
                    st.write("**Compliance Requirements:**")
                    for req in policy_analysis.get('compliance_requirements', []):
                        st.write(f"• {req}")
                
                st.write(f"**Source:** {announcement.get('url', 'Unknown')}")
    
    else:
        st.info("No policy analyses available yet. Run the tracker to begin collecting data.")

if __name__ == "__main__":
    create_dashboard()

class AnalysisQualityTracker:
    def __init__(self):
        self.quality_metrics = {
            'response_time': [],
            'analysis_completeness': [],
            'model_accuracy': []
        }
    
    def measure_analysis_quality(self, analysis: Dict, response_time: float):
        """Measure and track analysis quality metrics"""
        
        # Calculate completeness score
        required_fields = ['policy_changes', 'crypto_impact', 'affected_sectors']
        present_fields = sum(1 for field in required_fields if field in analysis)
        completeness = present_fields / len(required_fields)
        
        # Track metrics
        self.quality_metrics['response_time'].append(response_time)
        self.quality_metrics['analysis_completeness'].append(completeness)
        
        # Log quality metrics
        print(f"📊 Analysis Quality - Completeness: {completeness:.2f}, Time: {response_time:.2f}s")
        
        return {
            'completeness': completeness,
            'response_time': response_time,
            'quality_score': completeness * (1 / (1 + response_time))  # Combined score
        }

FROM python:3.11-slim

# Install system dependencies
RUN apt-get update && apt-get install -y \
    curl \
    git \
    && rm -rf /var/lib/apt/lists/*

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

# Set working directory
WORKDIR /app

# Copy requirements
COPY requirements.txt .
RUN pip install -r requirements.txt

# Copy application code
COPY . .

# Create data directory
RUN mkdir -p /app/data

# Expose port for dashboard
EXPOSE 8501

# Start script
COPY start.sh .
RUN chmod +x start.sh

CMD ["./start.sh"]

#!/bin/bash
# start.sh

echo "🚀 Starting Paul Atkins SEC Policy Tracker..."

# Start Ollama service
ollama serve &

# Wait for Ollama to be ready
sleep 10

# Pull required models
ollama pull deepseek-r1:7b
ollama pull llama3.3:70b
ollama pull qwen2.5:32b

# Start the policy tracker
python tracker.py &

# Start the dashboard
streamlit run dashboard.py --server.port 8501 --server.address 0.0.0.0

wait