Penny Stock Screening with Ollama: AI-Powered Risk Assessment and Opportunity Detection

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

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

pip install requests pandas yfinance beautifulsoup4 numpy

# config.py
import os

class Config:
    # API endpoints
    OLLAMA_BASE_URL = "http://localhost:11434"
    
    # Stock screening parameters
    MAX_PRICE = 5.00  # Penny stock threshold
    MIN_VOLUME = 100000  # Daily volume minimum
    MIN_MARKET_CAP = 1000000  # $1M minimum market cap
    
    # Risk assessment weights
    RISK_WEIGHTS = {
        'financial_health': 0.3,
        'management_quality': 0.2,
        'market_position': 0.25,
        'growth_potential': 0.25
    }

# data_collector.py
import yfinance as yf
import pandas as pd
import requests
from typing import List, Dict

class StockDataCollector:
    def __init__(self, config):
        self.config = config
    
    def get_penny_stocks(self) -> List[str]:
        """Fetch list of stocks under $5"""
        # Example using a stock screener API or predefined list
        penny_stocks = [
            'SNDL', 'NOK', 'NAKD', 'CTRM', 'SHIP',
            'GNUS', 'IDEX', 'XSPA', 'AYTU', 'IBIO'
        ]
        return penny_stocks
    
    def collect_stock_data(self, symbol: str) -> Dict:
        """Collect comprehensive stock data"""
        try:
            ticker = yf.Ticker(symbol)
            
            # Basic stock info
            info = ticker.info
            
            # Historical data (6 months)
            hist = ticker.history(period="6mo")
            
            # Recent news
            news = ticker.news[:5] if ticker.news else []
            
            return {
                'symbol': symbol,
                'price': info.get('currentPrice', 0),
                'volume': info.get('volume', 0),
                'market_cap': info.get('marketCap', 0),
                'pe_ratio': info.get('forwardPE', 'N/A'),
                'debt_to_equity': info.get('debtToEquity', 'N/A'),
                'revenue_growth': info.get('revenueGrowth', 'N/A'),
                'news': news,
                'historical_data': hist
            }
        except Exception as e:
            print(f"Error collecting data for {symbol}: {e}")
            return None

# ollama_analyzer.py
import requests
import json
from typing import Dict, List

class OllamaAnalyzer:
    def __init__(self, base_url="http://localhost:11434"):
        self.base_url = base_url
        self.model = "llama2:13b"
    
    def analyze_financial_health(self, stock_data: Dict) -> Dict:
        """Analyze financial health using AI"""
        
        prompt = f"""
        Analyze the financial health of {stock_data['symbol']} based on:
        - Current Price: ${stock_data['price']}
        - Market Cap: {stock_data['market_cap']}
        - P/E Ratio: {stock_data['pe_ratio']}
        - Debt-to-Equity: {stock_data['debt_to_equity']}
        - Revenue Growth: {stock_data['revenue_growth']}
        
        Provide a risk score (1-10, where 10 is highest risk) and explanation.
        Focus on: liquidity concerns, debt levels, profitability trends.
        
        Format: Risk Score: X | Explanation: [your analysis]
        """
        
        return self._query_ollama(prompt)
    
    def analyze_news_sentiment(self, news_items: List) -> Dict:
        """Analyze news sentiment for market perception"""
        
        if not news_items:
            return {"sentiment_score": 5, "explanation": "No recent news available"}
        
        news_text = " ".join([item.get('title', '') for item in news_items[:3]])
        
        prompt = f"""
        Analyze the sentiment of these recent news headlines:
        {news_text}
        
        Rate sentiment from 1-10 (1=very negative, 10=very positive).
        Identify key positive and negative factors.
        
        Format: Sentiment Score: X | Key Factors: [list factors]
        """
        
        return self._query_ollama(prompt)
    
    def _query_ollama(self, prompt: str) -> Dict:
        """Send query to Ollama and parse response"""
        try:
            response = requests.post(
                f"{self.base_url}/api/generate",
                json={
                    "model": self.model,
                    "prompt": prompt,
                    "stream": False
                }
            )
            
            result = response.json()
            return self._parse_response(result['response'])
            
        except Exception as e:
            print(f"Ollama query error: {e}")
            return {"score": 5, "explanation": "Analysis unavailable"}
    
    def _parse_response(self, response: str) -> Dict:
        """Parse Ollama response into structured data"""
        try:
            # Extract score and explanation from formatted response
            lines = response.split('|')
            score_line = lines[0].strip()
            explanation = lines[1].strip() if len(lines) > 1 else ""
            
            # Extract numeric score
            score = int(''.join(filter(str.isdigit, score_line)))
            
            return {
                "score": score,
                "explanation": explanation.replace("Explanation:", "").strip()
            }
        except:
            return {"score": 5, "explanation": response[:200]}

# risk_assessor.py
import numpy as np
from typing import Dict, List

class PennyStockRiskAssessor:
    def __init__(self, ollama_analyzer, config):
        self.analyzer = ollama_analyzer
        self.config = config
    
    def assess_comprehensive_risk(self, stock_data: Dict) -> Dict:
        """Perform comprehensive risk assessment"""
        
        # Financial health analysis
        financial_risk = self.analyzer.analyze_financial_health(stock_data)
        
        # News sentiment analysis
        sentiment_risk = self.analyzer.analyze_news_sentiment(stock_data['news'])
        
        # Technical analysis
        technical_risk = self._analyze_technical_indicators(stock_data)
        
        # Volume and liquidity risk
        liquidity_risk = self._assess_liquidity_risk(stock_data)
        
        # Calculate weighted overall risk
        overall_risk = self._calculate_weighted_risk({
            'financial': financial_risk['score'],
            'sentiment': sentiment_risk['score'],
            'technical': technical_risk['score'],
            'liquidity': liquidity_risk['score']
        })
        
        return {
            'overall_risk_score': overall_risk,
            'financial_risk': financial_risk,
            'sentiment_risk': sentiment_risk,
            'technical_risk': technical_risk,
            'liquidity_risk': liquidity_risk,
            'recommendation': self._generate_recommendation(overall_risk)
        }
    
    def _analyze_technical_indicators(self, stock_data: Dict) -> Dict:
        """Analyze technical indicators for risk assessment"""
        hist_data = stock_data['historical_data']
        
        if hist_data.empty:
            return {"score": 8, "explanation": "Insufficient historical data"}
        
        # Calculate volatility
        returns = hist_data['Close'].pct_change().dropna()
        volatility = returns.std() * np.sqrt(252)  # Annualized volatility
        
        # Price trend analysis
        recent_trend = (hist_data['Close'][-1] - hist_data['Close'][-30]) / hist_data['Close'][-30]
        
        # Risk scoring based on volatility and trend
        volatility_risk = min(10, volatility * 10)  # Scale volatility to 1-10
        trend_risk = 5 - (recent_trend * 10)  # Negative trend increases risk
        
        technical_score = (volatility_risk + trend_risk) / 2
        
        return {
            "score": round(technical_score, 1),
            "explanation": f"Volatility: {volatility:.2%}, Recent trend: {recent_trend:.2%}"
        }
    
    def _assess_liquidity_risk(self, stock_data: Dict) -> Dict:
        """Assess liquidity risk based on volume and market cap"""
        volume = stock_data.get('volume', 0)
        market_cap = stock_data.get('market_cap', 0)
        
        # Volume risk (lower volume = higher risk)
        volume_risk = 10 - min(10, volume / 100000)  # Scale to 1-10
        
        # Market cap risk (smaller cap = higher risk)
        cap_risk = 10 - min(10, market_cap / 1000000)  # Scale to 1-10
        
        liquidity_score = (volume_risk + cap_risk) / 2
        
        return {
            "score": round(liquidity_score, 1),
            "explanation": f"Daily volume: {volume:,}, Market cap: ${market_cap:,}"
        }
    
    def _calculate_weighted_risk(self, risk_scores: Dict) -> float:
        """Calculate weighted overall risk score"""
        weights = self.config.RISK_WEIGHTS
        
        weighted_score = (
            risk_scores['financial'] * weights['financial_health'] +
            risk_scores['sentiment'] * weights['management_quality'] +
            risk_scores['technical'] * weights['market_position'] +
            risk_scores['liquidity'] * weights['growth_potential']
        )
        
        return round(weighted_score, 1)
    
    def _generate_recommendation(self, risk_score: float) -> str:
        """Generate investment recommendation based on risk score"""
        if risk_score <= 3:
            return "LOW RISK: Consider for investment"
        elif risk_score <= 6:
            return "MODERATE RISK: Proceed with caution"
        elif risk_score <= 8:
            return "HIGH RISK: Avoid unless high risk tolerance"
        else:
            return "VERY HIGH RISK: Not recommended"

# opportunity_detector.py
from typing import Dict, List
import pandas as pd

class OpportunityDetector:
    def __init__(self, ollama_analyzer):
        self.analyzer = ollama_analyzer
    
    def detect_opportunities(self, stock_data: Dict) -> Dict:
        """Detect investment opportunities using AI analysis"""
        
        # Growth potential analysis
        growth_opportunity = self._analyze_growth_potential(stock_data)
        
        # Undervaluation detection
        value_opportunity = self._detect_undervaluation(stock_data)
        
        # Catalyst identification
        catalyst_opportunity = self._identify_catalysts(stock_data)
        
        # Technical opportunity
        technical_opportunity = self._detect_technical_opportunities(stock_data)
        
        # Overall opportunity score
        opportunity_score = self._calculate_opportunity_score({
            'growth': growth_opportunity['score'],
            'value': value_opportunity['score'],
            'catalyst': catalyst_opportunity['score'],
            'technical': technical_opportunity['score']
        })
        
        return {
            'opportunity_score': opportunity_score,
            'growth_potential': growth_opportunity,
            'value_opportunity': value_opportunity,
            'catalyst_potential': catalyst_opportunity,
            'technical_opportunity': technical_opportunity,
            'action_items': self._generate_action_items(opportunity_score)
        }
    
    def _analyze_growth_potential(self, stock_data: Dict) -> Dict:
        """Analyze growth potential using AI"""
        
        revenue_growth = stock_data.get('revenue_growth', 'N/A')
        market_cap = stock_data.get('market_cap', 0)
        
        prompt = f"""
        Analyze growth potential for {stock_data['symbol']}:
        - Revenue Growth: {revenue_growth}
        - Market Cap: ${market_cap:,}
        - Industry: Analyze based on current market trends
        
        Rate growth potential 1-10 (10=highest potential).
        Consider: market expansion, competitive advantages, scalability.
        
        Format: Growth Score: X | Potential: [your analysis]
        """
        
        return self.analyzer._query_ollama(prompt)
    
    def _detect_undervaluation(self, stock_data: Dict) -> Dict:
        """Detect potential undervaluation"""
        
        pe_ratio = stock_data.get('pe_ratio', 'N/A')
        price = stock_data.get('price', 0)
        
        prompt = f"""
        Assess if {stock_data['symbol']} is undervalued:
        - Current Price: ${price}
        - P/E Ratio: {pe_ratio}
        - Compare to industry averages
        
        Rate undervaluation potential 1-10 (10=significantly undervalued).
        Consider: financial metrics, market conditions, peer comparison.
        
        Format: Value Score: X | Analysis: [your assessment]
        """
        
        return self.analyzer._query_ollama(prompt)
    
    def _identify_catalysts(self, stock_data: Dict) -> Dict:
        """Identify potential price catalysts"""
        
        news_headlines = [item.get('title', '') for item in stock_data['news'][:3]]
        
        prompt = f"""
        Identify potential catalysts for {stock_data['symbol']}:
        Recent news: {' | '.join(news_headlines)}
        
        Look for: earnings releases, product launches, partnerships, regulatory approvals.
        Rate catalyst potential 1-10 (10=strong upcoming catalysts).
        
        Format: Catalyst Score: X | Catalysts: [list identified catalysts]
        """
        
        return self.analyzer._query_ollama(prompt)
    
    def _detect_technical_opportunities(self, stock_data: Dict) -> Dict:
        """Detect technical trading opportunities"""
        
        hist_data = stock_data['historical_data']
        
        if hist_data.empty:
            return {"score": 5, "explanation": "Insufficient data for technical analysis"}
        
        # Simple technical indicators
        current_price = hist_data['Close'][-1]
        ma_20 = hist_data['Close'][-20:].mean()
        ma_50 = hist_data['Close'][-50:].mean() if len(hist_data) >= 50 else ma_20
        
        # Support/resistance levels
        recent_high = hist_data['High'][-30:].max()
        recent_low = hist_data['Low'][-30:].min()
        
        # Technical scoring
        ma_signal = 7 if current_price > ma_20 > ma_50 else 3
        support_signal = 8 if current_price > (recent_low * 1.05) else 4
        
        technical_score = (ma_signal + support_signal) / 2
        
        return {
            "score": round(technical_score, 1),
            "explanation": f"Price vs MA20: {((current_price/ma_20-1)*100):.1f}%, Support level: ${recent_low:.2f}"
        }
    
    def _calculate_opportunity_score(self, scores: Dict) -> float:
        """Calculate overall opportunity score"""
        weights = {
            'growth': 0.3,
            'value': 0.25,
            'catalyst': 0.25,
            'technical': 0.2
        }
        
        opportunity_score = sum(scores[key] * weights[key] for key in scores)
        return round(opportunity_score, 1)
    
    def _generate_action_items(self, opportunity_score: float) -> List[str]:
        """Generate actionable recommendations"""
        if opportunity_score >= 7:
            return [
                "Research further - high opportunity potential",
                "Monitor entry points for position building",
                "Set up price alerts for key levels"
            ]
        elif opportunity_score >= 5:
            return [
                "Continue monitoring for improvements",
                "Wait for better entry opportunity",
                "Watch for catalyst developments"
            ]
        else:
            return [
                "Low opportunity - consider alternatives",
                "Focus on higher-scoring stocks",
                "Revisit in 30 days"
            ]

# main_screener.py
import pandas as pd
from config import Config
from data_collector import StockDataCollector
from ollama_analyzer import OllamaAnalyzer
from risk_assessor import PennyStockRiskAssessor
from opportunity_detector import OpportunityDetector

class PennyStockScreener:
    def __init__(self):
        self.config = Config()
        self.data_collector = StockDataCollector(self.config)
        self.analyzer = OllamaAnalyzer()
        self.risk_assessor = PennyStockRiskAssessor(self.analyzer, self.config)
        self.opportunity_detector = OpportunityDetector(self.analyzer)
    
    def run_complete_screening(self) -> pd.DataFrame:
        """Run complete penny stock screening process"""
        
        print("Starting penny stock screening with Ollama...")
        
        # Get penny stock list
        penny_stocks = self.data_collector.get_penny_stocks()
        results = []
        
        for symbol in penny_stocks:
            print(f"Analyzing {symbol}...")
            
            # Collect stock data
            stock_data = self.data_collector.collect_stock_data(symbol)
            if not stock_data:
                continue
            
            # Perform risk assessment
            risk_analysis = self.risk_assessor.assess_comprehensive_risk(stock_data)
            
            # Detect opportunities
            opportunity_analysis = self.opportunity_detector.detect_opportunities(stock_data)
            
            # Compile results
            result = {
                'symbol': symbol,
                'price': stock_data['price'],
                'market_cap': stock_data['market_cap'],
                'risk_score': risk_analysis['overall_risk_score'],
                'opportunity_score': opportunity_analysis['opportunity_score'],
                'recommendation': risk_analysis['recommendation'],
                'action_items': ', '.join(opportunity_analysis['action_items'][:2])
            }
            
            results.append(result)
        
        # Create DataFrame and sort by opportunity score
        df = pd.DataFrame(results)
        df = df.sort_values('opportunity_score', ascending=False)
        
        return df
    
    def generate_report(self, results_df: pd.DataFrame) -> str:
        """Generate summary report"""
        
        report = f"""
        PENNY STOCK SCREENING REPORT
        ============================
        
        Total Stocks Analyzed: {len(results_df)}
        
        TOP OPPORTUNITIES:
        {results_df.head(3)[['symbol', 'price', 'opportunity_score', 'risk_score']].to_string(index=False)}
        
        HIGH RISK ALERTS:
        {results_df[results_df['risk_score'] > 7][['symbol', 'risk_score', 'recommendation']].to_string(index=False)}
        
        RECOMMENDED ACTIONS:
        - Focus on stocks with opportunity_score > 6 and risk_score < 6
        - Investigate top 3 opportunities further
        - Set up monitoring for moderate-risk opportunities
        """
        
        return report

# Usage example
if __name__ == "__main__":
    screener = PennyStockScreener()
    results = screener.run_complete_screening()
    report = screener.generate_report(results)
    
    print(report)
    
    # Save results
    results.to_csv('penny_stock_screening_results.csv', index=False)

# alert_system.py
import smtplib
from email.mime.text import MimeText
from typing import Dict, List

class StockAlertSystem:
    def __init__(self, email_config: Dict = None):
        self.email_config = email_config
    
    def create_opportunity_alerts(self, screening_results: pd.DataFrame) -> List[Dict]:
        """Create alerts for high-opportunity stocks"""
        
        alerts = []
        
        # High opportunity, low risk stocks
        prime_opportunities = screening_results[
            (screening_results['opportunity_score'] > 7) & 
            (screening_results['risk_score'] < 5)
        ]
        
        for _, stock in prime_opportunities.iterrows():
            alerts.append({
                'type': 'HIGH_OPPORTUNITY',
                'symbol': stock['symbol'],
                'message': f"{stock['symbol']}: High opportunity (score: {stock['opportunity_score']}) with manageable risk (score: {stock['risk_score']})",
                'priority': 'HIGH'
            })
        
        # Sudden opportunity improvements
        # This would require historical comparison
        
        return alerts
    
    def send_email_alert(self, alert: Dict):
        """Send email alert for important opportunities"""
        if not self.email_config:
            print(f"ALERT: {alert['message']}")
            return
        
        # Email sending logic here
        pass

# batch_processor.py
import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor

class BatchStockProcessor:
    def __init__(self, max_workers=5):
        self.max_workers = max_workers
    
    async def process_stocks_batch(self, stock_symbols: List[str]) -> List[Dict]:
        """Process multiple stocks concurrently"""
        
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            loop = asyncio.get_event_loop()
            
            tasks = [
                loop.run_in_executor(
                    executor, 
                    self._process_single_stock, 
                    symbol
                ) for symbol in stock_symbols
            ]
            
            results = await asyncio.gather(*tasks)
            return [r for r in results if r is not None]
    
    def _process_single_stock(self, symbol: str) -> Dict:
        """Process single stock (placeholder for actual processing)"""
        # Your existing stock processing logic here
        pass

# troubleshooting.py
import requests
import time

def test_ollama_connection():
    """Test Ollama connection and model availability"""
    try:
        response = requests.get("http://localhost:11434/api/tags")
        if response.status_code == 200:
            models = response.json()
            print(f"Ollama connected. Available models: {[m['name'] for m in models['models']]}")
            return True
        else:
            print("Ollama not responding correctly")
            return False
    except requests.exceptions.ConnectionError:
        print("Cannot connect to Ollama. Is it running?")
        return False

def wait_for_ollama(max_wait=60):
    """Wait for Ollama to be ready"""
    for i in range(max_wait):
        if test_ollama_connection():
            return True
        print(f"Waiting for Ollama... ({i+1}/{max_wait})")
        time.sleep(1)
    return False