🔬 Inside the Momentum Scoring Algorithm

 1# Core momentum analyzer class structure
 2class MomentumAnalyzer:
 3    def __init__(self, csv_file_path):
 4        self.csv_file = csv_file_path
 5        self.data = None
 6        self.momentum_factors = {
 7            'ema_crossover': 0.25,      # 25% weight
 8            'price_vs_200ema': 0.20,    # 20% weight
 9            'atr_expansion': 0.15,      # 15% weight
10            'cci_momentum': 0.15,       # 15% weight
11            'volume_surge': 0.10,       # 10% weight
12            'price_position': 0.10,     # 10% weight
13            'velocity': 0.03,           # 3% weight
14            'breakout_confirm': 0.02    # 2% weight
15        }

 1# Exponential Moving Average calculation
 2def calculate_ema(prices, period):
 3    """
 4    Calculate Exponential Moving Average
 5    Formula: EMA = (Price * (2/(period+1))) + (Previous_EMA * (1-(2/(period+1))))
 6    """
 7    multiplier = 2 / (period + 1)
 8    ema_values = []
 9    
10    # Initialize with first price as first EMA
11    ema_values.append(prices[0])
12    
13    for i in range(1, len(prices)):
14        ema = (prices[i] * multiplier) + (ema_values[i-1] * (1 - multiplier))
15        ema_values.append(ema)
16    
17    return ema_values

 1# Volume surge detection and scoring
 2def analyze_volume_momentum(volumes, prices):
 3    """
 4    Analyze volume patterns for momentum confirmation
 5    Returns score based on recent volume vs historical average
 6    """
 7    volume_20ma = calculate_simple_ma(volumes, 20)
 8    recent_volume = volumes[-1]
 9    avg_volume = volume_20ma[-1]
10    
11    # Volume ratio calculation
12    volume_ratio = recent_volume / avg_volume
13    
14    # Score based on volume surge magnitude
15    if volume_ratio > 2.0:        # 2x average volume
16        volume_score = 100
17    elif volume_ratio > 1.5:      # 1.5x average volume
18        volume_score = 75
19    elif volume_ratio > 1.2:      # 1.2x average volume
20        volume_score = 50
21    else:
22        volume_score = 25
23    
24    return volume_score

 1# EMA Crossover calculation and scoring
 2def calculate_ema_crossover_score(ema20, ema50):
 3    """Calculate momentum score based on EMA crossover strength"""
 4    current_ema20 = ema20[-1]
 5    current_ema50 = ema50[-1]
 6    
 7    # Calculate percentage difference
 8    ema_diff_pct = ((current_ema20 - current_ema50) / current_ema50) * 100
 9    
10    # Score based on crossover strength
11    if ema_diff_pct >= 5.0:
12        return 100  # Strong bullish crossover
13    elif ema_diff_pct >= 2.0:
14        return 75   # Moderate bullish crossover
15    elif ema_diff_pct > 0:
16        return 50   # Weak bullish crossover
17    else:
18        return 0    # Bearish or no crossover

 1# ATR calculation and momentum scoring
 2def calculate_atr_momentum_score(highs, lows, closes, period=14):
 3    """
 4    Calculate ATR and score based on volatility expansion
 5    ATR = SMA of True Range over specified period
 6    """
 7    true_ranges = []
 8    
 9    for i in range(1, len(highs)):
10        tr1 = highs[i] - lows[i]                    # Current high-low
11        tr2 = abs(highs[i] - closes[i-1])          # High vs prev close
12        tr3 = abs(lows[i] - closes[i-1])           # Low vs prev close
13        
14        true_range = max(tr1, tr2, tr3)
15        true_ranges.append(true_range)
16    
17    # Calculate ATR as simple moving average
18    atr_values = calculate_simple_ma(true_ranges, period)
19    
20    # Compare current ATR to historical average
21    current_atr = atr_values[-1]
22    avg_atr = sum(atr_values[-30:]) / 30  # 30-day ATR average
23    
24    atr_ratio = current_atr / avg_atr
25    
26    # Score based on volatility expansion
27    if atr_ratio > 1.5:
28        return 100  # High volatility expansion
29    elif atr_ratio > 1.2:
30        return 75   # Moderate expansion
31    elif atr_ratio > 1.0:
32        return 50   # Normal volatility
33    else:
34        return 25   # Low volatility

 1# CCI calculation and momentum interpretation
 2def calculate_cci_momentum_score(highs, lows, closes, period=20):
 3    """
 4    Calculate Commodity Channel Index for momentum scoring
 5    CCI = (Typical Price - SMA of Typical Price) / (0.015 * Mean Deviation)
 6    """
 7    typical_prices = [(h + l + c) / 3 for h, l, c in zip(highs, lows, closes)]
 8    
 9    cci_values = []
10    for i in range(period - 1, len(typical_prices)):
11        tp_slice = typical_prices[i-period+1:i+1]
12        
13        # Calculate simple moving average of typical price
14        sma_tp = sum(tp_slice) / period
15        
16        # Calculate mean deviation
17        mean_deviation = sum([abs(tp - sma_tp) for tp in tp_slice]) / period
18        
19        # CCI formula
20        cci = (typical_prices[i] - sma_tp) / (0.015 * mean_deviation)
21        cci_values.append(cci)
22    
23    current_cci = cci_values[-1]
24    
25    # Score based on CCI momentum levels
26    if current_cci > 150:
27        return 100  # Strong momentum
28    elif current_cci > 100:
29        return 75   # Good momentum
30    elif current_cci > 50:
31        return 50   # Moderate momentum
32    else:
33        return 25   # Weak momentum

 1# Complete momentum score compilation
 2def compile_momentum_score(self, stock_data):
 3    """
 4    Calculate final weighted momentum score from all factors
 5    Returns score from 0-100 indicating momentum strength
 6    """
 7    
 8    # Calculate individual factor scores
 9    factor_scores = {
10        'ema_crossover': self.calculate_ema_crossover_score(stock_data),
11        'price_vs_200ema': self.calculate_price_vs_ema_score(stock_data),
12        'atr_expansion': self.calculate_atr_momentum_score(stock_data),
13        'cci_momentum': self.calculate_cci_momentum_score(stock_data),
14        'volume_surge': self.calculate_volume_score(stock_data),
15        'price_position': self.calculate_52week_position_score(stock_data),
16        'velocity': self.calculate_velocity_score(stock_data),
17        'breakout_confirm': self.calculate_breakout_score(stock_data)
18    }
19    
20    # Apply weights and calculate final score
21    weighted_score = 0
22    for factor, score in factor_scores.items():
23        weight = self.momentum_factors[factor]
24        weighted_score += score * weight
25    
26    # Normalize to 0-100 scale and round
27    final_score = round(weighted_score, 2)
28    
29    # Log factor breakdown for debugging
30    self.log_factor_breakdown(factor_scores, weighted_score)
31    
32    return final_score

 1# Efficient batch processing for large datasets
 2def process_batch_stocks(self, stock_symbols, batch_size=50):
 3    """
 4    Process stocks in batches to optimize memory usage
 5    and enable parallel computation
 6    """
 7    results = []
 8    
 9    for i in range(0, len(stock_symbols), batch_size):
10        batch = stock_symbols[i:i+batch_size]
11        
12        # Parallel processing using multiprocessing
13        with Pool(processes=cpu_count()) as pool:
14            batch_results = pool.map(self.analyze_single_stock, batch)
15        
16        results.extend(batch_results)
17        
18        # Progress tracking
19        progress = (i + batch_size) / len(stock_symbols) * 100
20        print(f"Processing: {progress:.1f}% complete")
21    
22    return results

 1# Sector-specific momentum adjustments
 2def apply_sector_adjustments(self, base_score, sector):
 3    """
 4    Apply sector-specific momentum adjustments
 5    Different sectors exhibit different momentum characteristics
 6    """
 7    sector_multipliers = {
 8        'Technology': 1.15,      # Tech stocks show stronger momentum
 9        'Banking': 0.95,         # Banks are more stable, less explosive
10        'Pharmaceuticals': 1.10, # Pharma can show strong moves
11        'FMCG': 0.90,            # FMCG typically less volatile
12        'Metals': 1.20,          # Metals highly momentum-driven
13        'Energy': 1.05           # Energy moderate momentum
14    }
15    
16    multiplier = sector_multipliers.get(sector, 1.0)
17    adjusted_score = base_score * multiplier
18    
19    # Cap at 100 to maintain scale
20    return min(adjusted_score, 100)

 1# Required Python libraries
 2import pandas as pd
 3import numpy as np
 4from datetime import datetime, timedelta
 5import multiprocessing
 6from typing import List, Dict, Tuple
 7
 8# Install required packages
 9# pip install pandas numpy yfinance requests beautifulsoup4

 1# Complete momentum analyzer template
 2class MomentumScoringEngine:
 3    def __init__(self):
 4        self.factor_weights = {
 5            'ema_crossover': 0.25,
 6            'price_vs_200ema': 0.20,
 7            'atr_expansion': 0.15,
 8            'cci_momentum': 0.15,
 9            'volume_surge': 0.10,
10            'price_position': 0.10,
11            'velocity': 0.03,
12            'breakout_confirm': 0.02
13        }
14    
15    def load_data(self, csv_path: str) -> pd.DataFrame:
16        """Load and validate NSE CSV data"""
17        pass
18    
19    def calculate_indicators(self, data: pd.DataFrame) -> Dict:
20        """Calculate all technical indicators"""
21        pass
22    
23    def score_momentum(self, indicators: Dict) -> float:
24        """Compile weighted momentum score"""
25        pass
26    
27    def analyze_stock(self, symbol: str) -> Dict:
28        """Complete analysis pipeline for single stock"""
29        pass

 1# Unit testing framework for algorithm validation
 2import unittest
 3
 4class TestMomentumAlgorithm(unittest.TestCase):
 5    def setUp(self):
 6        self.analyzer = MomentumScoringEngine()
 7        self.test_data = self.load_test_data()
 8    
 9    def test_ema_calculation(self):
10        """Test EMA calculation accuracy"""
11        test_prices = [100, 102, 101, 103, 105]
12        ema = self.analyzer.calculate_ema(test_prices, 3)
13        expected = 102.75  # Known EMA value
14        self.assertAlmostEqual(ema[-1], expected, places=2)
15    
16    def test_score_range(self):
17        """Ensure scores are in 0-100 range"""
18        for symbol in self.test_symbols:
19            score = self.analyzer.analyze_stock(symbol)
20            self.assertTrue(0 <= score <= 100)
21
22# Run tests
23if __name__ == '__main__':
24    unittest.main()