Build a Bitcoin Disclosure Tracker: Automated SEC Filing Analysis with Ollama AI

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

# Windows - Download from ollama.com/download

# Verify installation
ollama --version

# General purpose model for document analysis
ollama pull llama3.2:latest

# Bitcoin-specialized model for crypto-specific analysis
ollama pull shreyshah/satoshi-7b-q4_k_m

# Lightweight model for quick processing
ollama pull phi3:mini

# requirements.txt
requests==2.31.0
beautifulsoup4==4.12.2
pandas==2.0.3
python-dotenv==1.0.0
ollama==0.1.7
schedule==1.2.0

pip install -r requirements.txt

import requests
import json
import time
from datetime import datetime, timedelta
import pandas as pd

class SECFilingRetriever:
    def __init__(self):
        self.base_url = "https://data.sec.gov/submissions/"
        self.headers = {
            'User-Agent': 'Bitcoin-Tracker/1.0 (contact@yourcompany.com)'
        }
        self.rate_limit_delay = 0.1  # SEC requires 10 requests per second max
    
    def get_company_filings(self, cik):
        """Retrieve recent filings for a company by CIK number"""
        url = f"{self.base_url}CIK{cik:010d}.json"
        
        try:
            response = requests.get(url, headers=self.headers)
            response.raise_for_status()
            time.sleep(self.rate_limit_delay)
            
            return response.json()
        except requests.RequestException as e:
            print(f"Error retrieving filings for CIK {cik}: {e}")
            return None
    
    def filter_recent_filings(self, filings_data, days_back=90):
        """Filter filings to recent submissions only"""
        if not filings_data or 'filings' not in filings_data:
            return []
        
        cutoff_date = datetime.now() - timedelta(days=days_back)
        recent_filings = []
        
        filings = filings_data['filings']['recent']
        
        for i, filing_date in enumerate(filings['filingDate']):
            if datetime.strptime(filing_date, '%Y-%m-%d') >= cutoff_date:
                # Focus on forms likely to contain Bitcoin disclosures
                form_type = filings['form'][i]
                if form_type in ['10-K', '10-Q', '8-K', 'DEF 14A']:
                    recent_filings.append({
                        'accessionNumber': filings['accessionNumber'][i],
                        'filingDate': filing_date,
                        'form': form_type,
                        'primaryDocument': filings['primaryDocument'][i]
                    })
        
        return recent_filings

from bs4 import BeautifulSoup
import re

class SECDocumentProcessor:
    def __init__(self):
        self.headers = {
            'User-Agent': 'Bitcoin-Tracker/1.0 (contact@yourcompany.com)'
        }
    
    def extract_filing_text(self, accession_number, primary_document, cik):
        """Download and extract text from SEC filing"""
        # Format accession number for URL (remove hyphens)
        acc_no_formatted = accession_number.replace('-', '')
        
        # Construct document URL
        doc_url = f"https://www.sec.gov/Archives/edgar/data/{cik}/{acc_no_formatted}/{primary_document}"
        
        try:
            response = requests.get(doc_url, headers=self.headers)
            response.raise_for_status()
            
            # Parse HTML content
            soup = BeautifulSoup(response.content, 'html.parser')
            
            # Extract text and clean formatting
            text = soup.get_text()
            cleaned_text = self.clean_text(text)
            
            return cleaned_text
            
        except requests.RequestException as e:
            print(f"Error downloading document {primary_document}: {e}")
            return None
    
    def clean_text(self, text):
        """Clean and normalize extracted text"""
        # Remove extra whitespace and normalize line breaks
        text = re.sub(r'\s+', ' ', text)
        
        # Remove SEC-specific formatting artifacts
        text = re.sub(r'Table of Contents', '', text, flags=re.IGNORECASE)
        text = re.sub(r'<[^>]+>', '', text)  # Remove any remaining HTML tags
        
        return text.strip()
    
    def extract_relevant_sections(self, text):
        """Extract sections likely to contain Bitcoin disclosure"""
        sections = {}
        
        # Define section patterns to search for
        section_patterns = {
            'balance_sheet': r'(consolidated )?balance sheets?.*?(?=\n\s*\n|\Z)',
            'notes': r'notes? to.*?financial statements.*?(?=\n\s*\n|\Z)', 
            'md_a': r'management.?s discussion and analysis.*?(?=\n\s*\n|\Z)',
            'risk_factors': r'risk factors.*?(?=\n\s*\n|\Z)',
            'liquidity': r'liquidity and capital resources.*?(?=\n\s*\n|\Z)'
        }
        
        for section_name, pattern in section_patterns.items():
            matches = re.findall(pattern, text, re.IGNORECASE | re.DOTALL)
            if matches:
                sections[section_name] = matches[0][:5000]  # Limit text length
        
        return sections

import ollama

class BitcoinDisclosureAnalyzer:
    def __init__(self, model_name="llama3.2:latest"):
        self.model_name = model_name
        self.bitcoin_keywords = [
            'bitcoin', 'btc', 'cryptocurrency', 'crypto asset', 
            'digital asset', 'virtual currency', 'blockchain'
        ]
    
    def analyze_document_for_bitcoin(self, text, company_name):
        """Analyze document text for Bitcoin-related disclosures"""
        
        # First, check if document contains Bitcoin references
        if not self.contains_bitcoin_references(text):
            return {
                'has_bitcoin_disclosure': False,
                'confidence': 0.0,
                'details': None
            }
        
        # Extract relevant sections for detailed analysis
        prompt = self.create_analysis_prompt(text, company_name)
        
        try:
            response = ollama.chat(
                model=self.model_name,
                messages=[
                    {
                        'role': 'system',
                        'content': 'You are a financial analyst specializing in cryptocurrency disclosures in SEC filings. Provide accurate, specific analysis of Bitcoin holdings and related financial information.'
                    },
                    {
                        'role': 'user', 
                        'content': prompt
                    }
                ]
            )
            
            analysis_result = self.parse_analysis_response(response['message']['content'])
            return analysis_result
            
        except Exception as e:
            print(f"Error during Ollama analysis: {e}")
            return {
                'has_bitcoin_disclosure': False,
                'confidence': 0.0,
                'error': str(e)
            }
    
    def contains_bitcoin_references(self, text):
        """Quick check for Bitcoin-related keywords"""
        text_lower = text.lower()
        return any(keyword in text_lower for keyword in self.bitcoin_keywords)
    
    def create_analysis_prompt(self, text, company_name):
        """Create a structured prompt for Bitcoin disclosure analysis"""
        
        # Truncate text to manage token limits
        text_excerpt = text[:4000] if len(text) > 4000 else text
        
        prompt = f"""
Analyze the following SEC filing excerpt from {company_name} for Bitcoin or cryptocurrency disclosures.

DOCUMENT EXCERPT:
{text_excerpt}

Please provide analysis in this exact JSON format:
{{
    "has_bitcoin_disclosure": true/false,
    "bitcoin_amount": "number of BTC or null",
    "fiat_value": "USD value or null", 
    "acquisition_cost": "original purchase price or null",
    "unrealized_gain_loss": "gain/loss amount or null",
    "disclosure_type": "balance sheet/transaction/risk factor/other",
    "confidence_score": 0.0-1.0,
    "key_quotes": ["relevant text excerpts"],
    "analysis_notes": "additional context"
}}

Focus on:
1. Specific Bitcoin/cryptocurrency amounts or values
2. Changes in holdings (purchases, sales, impairments)
3. Accounting treatment and valuation methods
4. Risk disclosures related to cryptocurrency investments
5. Material impacts on financial statements

Only mark has_bitcoin_disclosure as true if there is actual financial disclosure, not just mentions of accepting crypto payments or general risk factors.
"""
        return prompt
    
    def parse_analysis_response(self, response_text):
        """Parse and validate the Ollama response"""
        try:
            # Extract JSON from response if it's embedded in text
            json_match = re.search(r'\{.*\}', response_text, re.DOTALL)
            if json_match:
                json_str = json_match.group()
                analysis = json.loads(json_str)
                
                # Validate required fields
                required_fields = ['has_bitcoin_disclosure', 'confidence_score']
                for field in required_fields:
                    if field not in analysis:
                        analysis[field] = False if field == 'has_bitcoin_disclosure' else 0.0
                
                return analysis
            else:
                # Fallback parsing for non-JSON responses
                return self.fallback_parse(response_text)
                
        except json.JSONDecodeError:
            return self.fallback_parse(response_text)
    
    def fallback_parse(self, response_text):
        """Fallback parsing when JSON parsing fails"""
        has_disclosure = any(word in response_text.lower() for word in 
                           ['bitcoin', 'btc', 'cryptocurrency', 'digital asset'])
        
        confidence = 0.5 if has_disclosure else 0.1
        
        return {
            'has_bitcoin_disclosure': has_disclosure,
            'confidence_score': confidence,
            'analysis_notes': response_text[:500],
            'parsing_method': 'fallback'
        }

import json
from datetime import datetime
import logging

class BitcoinDisclosureTracker:
    def __init__(self, model_name="llama3.2:latest"):
        self.retriever = SECFilingRetriever()
        self.processor = SECDocumentProcessor()
        self.analyzer = BitcoinDisclosureAnalyzer(model_name)
        
        # Known Bitcoin-holding companies (CIK numbers)
        self.target_companies = {
            'Tesla': '0001318605',
            'MicroStrategy': '0001050446', 
            'Coinbase': '0001679788',
            'Block': '0001512673',
            'Marathon Digital': '0001507605',
            'Riot Platforms': '0001167419',
            'Galaxy Digital': '0001808682'
        }
        
        # Setup logging
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)
    
    def scan_company(self, company_name, cik, days_back=30):
        """Scan a single company for Bitcoin disclosures"""
        self.logger.info(f"Scanning {company_name} (CIK: {cik})...")
        
        # Get recent filings
        filings_data = self.retriever.get_company_filings(int(cik))
        if not filings_data:
            return None
        
        recent_filings = self.retriever.filter_recent_filings(filings_data, days_back)
        
        results = []
        for filing in recent_filings:
            self.logger.info(f"Processing {filing['form']} filed {filing['filingDate']}")
            
            # Extract document text
            text = self.processor.extract_filing_text(
                filing['accessionNumber'], 
                filing['primaryDocument'], 
                cik.lstrip('0')
            )
            
            if text:
                # Analyze for Bitcoin disclosures
                analysis = self.analyzer.analyze_document_for_bitcoin(text, company_name)
                
                if analysis['has_bitcoin_disclosure'] and analysis['confidence_score'] > 0.6:
                    result = {
                        'company': company_name,
                        'cik': cik,
                        'filing_date': filing['filingDate'],
                        'form_type': filing['form'],
                        'accession_number': filing['accessionNumber'],
                        'analysis': analysis,
                        'scan_timestamp': datetime.now().isoformat()
                    }
                    results.append(result)
                    
                    self.logger.info(f"✓ Bitcoin disclosure found in {filing['form']}")
        
        return results
    
    def scan_all_companies(self, days_back=30):
        """Scan all target companies for Bitcoin disclosures"""
        all_results = []
        
        for company_name, cik in self.target_companies.items():
            try:
                company_results = self.scan_company(company_name, cik, days_back)
                if company_results:
                    all_results.extend(company_results)
                
                # Rate limiting - respect SEC guidelines
                time.sleep(0.2)
                
            except Exception as e:
                self.logger.error(f"Error scanning {company_name}: {e}")
        
        return all_results
    
    def generate_report(self, results, output_file="bitcoin_disclosure_report.json"):
        """Generate a comprehensive disclosure report"""
        report = {
            'scan_date': datetime.now().isoformat(),
            'total_disclosures_found': len(results),
            'companies_with_disclosures': len(set(r['company'] for r in results)),
            'disclosures': results,
            'summary': self.create_summary(results)
        }
        
        # Save to file
        with open(output_file, 'w') as f:
            json.dump(report, f, indent=2)
        
        self.logger.info(f"Report saved to {output_file}")
        return report
    
    def create_summary(self, results):
        """Create a summary of disclosure findings"""
        summary = {
            'by_company': {},
            'by_form_type': {},
            'recent_activity': []
        }
        
        for result in results:
            company = result['company']
            form_type = result['form_type']
            
            # Company summary
            if company not in summary['by_company']:
                summary['by_company'][company] = 0
            summary['by_company'][company] += 1
            
            # Form type summary  
            if form_type not in summary['by_form_type']:
                summary['by_form_type'][form_type] = 0
            summary['by_form_type'][form_type] += 1
            
            # Recent activity (last 7 days)
            filing_date = datetime.strptime(result['filing_date'], '%Y-%m-%d')
            if (datetime.now() - filing_date).days <= 7:
                summary['recent_activity'].append({
                    'company': company,
                    'form': form_type,
                    'date': result['filing_date']
                })
        
        return summary

# Initialize the tracker
tracker = BitcoinDisclosureTracker(model_name="llama3.2:latest")

# Scan all companies for disclosures in the last 30 days
results = tracker.scan_all_companies(days_back=30)

# Generate and save report
report = tracker.generate_report(results)

print(f"Found {len(results)} Bitcoin disclosures")
print(f"Report saved with {report['total_disclosures_found']} findings")

import schedule
import time

def daily_scan():
    """Daily automated scan function"""
    tracker = BitcoinDisclosureTracker()
    
    # Scan for new filings from last 2 days
    results = tracker.scan_all_companies(days_back=2)
    
    if results:
        timestamp = datetime.now().strftime("%Y%m%d")
        report = tracker.generate_report(results, f"daily_report_{timestamp}.json")
        print(f"Daily scan complete: {len(results)} new disclosures found")
    else:
        print("No new Bitcoin disclosures found")

# Schedule daily scans at 6 AM
schedule.every().day.at("06:00").do(daily_scan)

# Run scheduler
while True:
    schedule.run_pending()
    time.sleep(3600)  # Check every hour

class AdvancedBitcoinAnalyzer:
    def __init__(self):
        self.models = {
            'extraction': 'llama3.2:latest',       # General text analysis
            'bitcoin_specific': 'shreyshah/satoshi-7b-q4_k_m',  # Bitcoin expertise
            'financial': 'phi3:mini'               # Quick financial calculations
        }
    
    def multi_model_analysis(self, text, company_name):
        """Run analysis with multiple specialized models"""
        results = {}
        
        # Quick extraction with lightweight model
        extraction_result = self.analyze_with_model(
            text, company_name, self.models['financial'], 
            "Extract specific Bitcoin amounts and values mentioned"
        )
        results['extraction'] = extraction_result
        
        # Detailed Bitcoin analysis with specialized model
        bitcoin_result = self.analyze_with_model(
            text, company_name, self.models['bitcoin_specific'],
            "Analyze Bitcoin strategy and holdings from a cryptocurrency expert perspective"
        )
        results['bitcoin_analysis'] = bitcoin_result
        
        # Comprehensive financial analysis
        financial_result = self.analyze_with_model(
            text, company_name, self.models['extraction'],
            "Provide detailed financial statement analysis of cryptocurrency holdings"
        )
        results['financial_analysis'] = financial_result
        
        return self.combine_analysis_results(results)
    
    def analyze_with_model(self, text, company_name, model, focus):
        """Run analysis with a specific model"""
        prompt = f"""
        {focus}
        
        Company: {company_name}
        Document excerpt: {text[:3000]}
        
        Focus on accuracy and specific details.
        """
        
        try:
            response = ollama.chat(
                model=model,
                messages=[{'role': 'user', 'content': prompt}]
            )
            return response['message']['content']
        except Exception as e:
            return f"Error with {model}: {e}"

class PerformanceMonitor:
    def __init__(self):
        self.metrics = {
            'total_documents_processed': 0,
            'bitcoin_disclosures_found': 0,
            'processing_times': [],
            'model_accuracy': {},
            'false_positives': 0,
            'false_negatives': 0
        }
    
    def log_analysis(self, start_time, end_time, result, model_used):
        """Log analysis performance metrics"""
        processing_time = end_time - start_time
        
        self.metrics['total_documents_processed'] += 1
        self.metrics['processing_times'].append(processing_time)
        
        if result['has_bitcoin_disclosure']:
            self.metrics['bitcoin_disclosures_found'] += 1
        
        # Track model performance
        if model_used not in self.metrics['model_accuracy']:
            self.metrics['model_accuracy'][model_used] = {
                'total_analyses': 0,
                'high_confidence': 0,
                'avg_confidence': 0.0
            }
        
        model_stats = self.metrics['model_accuracy'][model_used]
        model_stats['total_analyses'] += 1
        
        if result['confidence_score'] > 0.8:
            model_stats['high_confidence'] += 1
    
    def generate_performance_report(self):
        """Generate performance analysis report"""
        avg_processing_time = sum(self.metrics['processing_times']) / len(self.metrics['processing_times']) if self.metrics['processing_times'] else 0
        
        report = {
            'total_documents': self.metrics['total_documents_processed'],
            'bitcoin_disclosures_found': self.metrics['bitcoin_disclosures_found'],
            'detection_rate': self.metrics['bitcoin_disclosures_found'] / max(1, self.metrics['total_documents_processed']),
            'avg_processing_time_seconds': avg_processing_time,
            'model_performance': self.metrics['model_accuracy']
        }
        
        return report

from flask import Flask, render_template, jsonify
import json
from datetime import datetime, timedelta

app = Flask(__name__)

class BitcoinDashboard:
    def __init__(self, data_file="bitcoin_disclosure_report.json"):
        self.data_file = data_file
    
    def load_latest_data(self):
        """Load the most recent disclosure data"""
        try:
            with open(self.data_file, 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            return {'disclosures': [], 'summary': {}}
    
    def get_dashboard_data(self):
        """Prepare data for dashboard display"""
        data = self.load_latest_data()
        
        dashboard_data = {
            'last_update': data.get('scan_date', 'Never'),
            'total_disclosures': data.get('total_disclosures_found', 0),
            'companies_tracked': len(data.get('summary', {}).get('by_company', {})),
            'recent_disclosures': self.format_recent_disclosures(data.get('disclosures', [])),
            'company_summary': data.get('summary', {}).get('by_company', {}),
            'form_type_breakdown': data.get('summary', {}).get('by_form_type', {})
        }
        
        return dashboard_data
    
    def format_recent_disclosures(self, disclosures):
        """Format disclosures for display"""
        # Sort by filing date, most recent first
        sorted_disclosures = sorted(
            disclosures, 
            key=lambda x: x['filing_date'], 
            reverse=True
        )
        
        formatted = []
        for disclosure in sorted_disclosures[:10]:  # Show last 10
            formatted.append({
                'company': disclosure['company'],
                'form_type': disclosure['form_type'],
                'filing_date': disclosure['filing_date'],
                'confidence': disclosure['analysis']['confidence_score'],
                'bitcoin_amount': disclosure['analysis'].get('bitcoin_amount', 'Not specified'),
                'fiat_value': disclosure['analysis'].get('fiat_value', 'Not specified')
            })
        
        return formatted

dashboard = BitcoinDashboard()

@app.route('/')
def index():
    """Main dashboard page"""
    data = dashboard.get_dashboard_data()
    return render_template('dashboard.html', data=data)

@app.route('/api/data')
def api_data():
    """API endpoint for dashboard data"""
    return jsonify(dashboard.get_dashboard_data())

if __name__ == '__main__':
    app.run(debug=True, port=5000)

# Dockerfile
FROM python:3.11-slim

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

# Set working directory
WORKDIR /app

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

# Copy application code
COPY . .

# Download Ollama models
RUN ollama serve & sleep 10 && \
    ollama pull llama3.2:latest && \
    ollama pull phi3:mini

# Expose port for dashboard
EXPOSE 5000

# Run the application
CMD ["python", "main.py"]

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

class Config:
    # SEC API settings
    SEC_USER_AGENT = os.getenv('SEC_USER_AGENT', 'Bitcoin-Tracker/1.0')
    SEC_RATE_LIMIT = float(os.getenv('SEC_RATE_LIMIT', '0.1'))
    
    # Ollama settings
    OLLAMA_MODEL = os.getenv('OLLAMA_MODEL', 'llama3.2:latest')
    OLLAMA_HOST = os.getenv('OLLAMA_HOST', 'localhost:11434')
    
    # Analysis settings
    CONFIDENCE_THRESHOLD = float(os.getenv('CONFIDENCE_THRESHOLD', '0.6'))
    MAX_TEXT_LENGTH = int(os.getenv('MAX_TEXT_LENGTH', '4000'))
    
    # Monitoring settings
    SCAN_INTERVAL_HOURS = int(os.getenv('SCAN_INTERVAL_HOURS', '24'))
    LOOKBACK_DAYS = int(os.getenv('LOOKBACK_DAYS', '30'))
    
    # Output settings
    REPORT_OUTPUT_DIR = os.getenv('REPORT_OUTPUT_DIR', './reports/')
    LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO')