Performance Tuning

# neural-engine/src/core/performance/batch_processor.py
from typing import List, Dict, Any
import numpy as np
from concurrent.futures import ThreadPoolExecutor
import asyncio
 
class BatchSignalProcessor:
    """Optimized batch processing for neural signals"""
 
    def __init__(self, batch_size: int = 1000, num_workers: int = 4):
        self.batch_size = batch_size
        self.num_workers = num_workers
        self.executor = ThreadPoolExecutor(max_workers=num_workers)
 
    async def process_batch(self, signals: List[np.ndarray]) -> List[Dict[str, Any]]:
        """Process signals in optimized batches"""
        # Batch signals for efficient processing
        batches = [
            signals[i:i + self.batch_size]
            for i in range(0, len(signals), self.batch_size)
        ]
 
        # Process batches in parallel
        loop = asyncio.get_event_loop()
        futures = [
            loop.run_in_executor(self.executor, self._process_batch_sync, batch)
            for batch in batches
        ]
 
        results = await asyncio.gather(*futures)
        return [item for batch in results for item in batch]
 
    def _process_batch_sync(self, batch: List[np.ndarray]) -> List[Dict[str, Any]]:
        """Synchronous batch processing with vectorization"""
        # Stack signals for vectorized operations
        stacked = np.stack(batch)
 
        # Vectorized FFT computation
        fft_results = np.fft.rfft(stacked, axis=1)
        power_spectrum = np.abs(fft_results) ** 2
 
        # Vectorized feature extraction
        features = {
            'mean_power': np.mean(power_spectrum, axis=1),
            'peak_frequency': np.argmax(power_spectrum, axis=1),
            'spectral_entropy': self._compute_spectral_entropy_vectorized(power_spectrum)
        }
 
        return [
            {
                'features': {k: v[i] for k, v in features.items()},
                'timestamp': np.datetime64('now')
            }
            for i in range(len(batch))
        ]
 
    def _compute_spectral_entropy_vectorized(self, power_spectrum: np.ndarray) -> np.ndarray:
        """Vectorized spectral entropy computation"""
        # Normalize power spectrum
        norm_ps = power_spectrum / np.sum(power_spectrum, axis=1, keepdims=True)
        # Compute entropy
        return -np.sum(norm_ps * np.log2(norm_ps + 1e-10), axis=1)

# neural-engine/src/core/performance/stream_processor.py
import asyncio
from typing import AsyncIterator, Dict, Any
import numpy as np
from collections import deque
 
class StreamSignalProcessor:
    """Optimized stream processing with sliding windows"""
 
    def __init__(self, window_size: int = 1000, overlap: float = 0.5):
        self.window_size = window_size
        self.overlap = overlap
        self.step_size = int(window_size * (1 - overlap))
        self.buffer = deque(maxlen=window_size)
 
    async def process_stream(
        self,
        signal_stream: AsyncIterator[np.ndarray]
    ) -> AsyncIterator[Dict[str, Any]]:
        """Process streaming signals with sliding window"""
        sample_count = 0
 
        async for sample in signal_stream:
            self.buffer.append(sample)
            sample_count += 1
 
            # Process window when we have enough samples
            if len(self.buffer) == self.window_size and \
               sample_count % self.step_size == 0:
 
                # Convert buffer to numpy array
                window = np.array(self.buffer)
 
                # Process window asynchronously
                result = await self._process_window_async(window)
                yield result
 
    async def _process_window_async(self, window: np.ndarray) -> Dict[str, Any]:
        """Asynchronous window processing"""
        # Run CPU-intensive operations in executor
        loop = asyncio.get_event_loop()
 
        # Parallel feature extraction
        features = await asyncio.gather(
            loop.run_in_executor(None, self._compute_time_features, window),
            loop.run_in_executor(None, self._compute_frequency_features, window),
            loop.run_in_executor(None, self._compute_nonlinear_features, window)
        )
 
        return {
            'time_features': features[0],
            'frequency_features': features[1],
            'nonlinear_features': features[2],
            'window_start': window[0],
            'window_end': window[-1]
        }
 
    def _compute_time_features(self, window: np.ndarray) -> Dict[str, float]:
        """Optimized time-domain feature extraction"""
        return {
            'mean': np.mean(window),
            'std': np.std(window),
            'skewness': self._fast_skewness(window),
            'kurtosis': self._fast_kurtosis(window)
        }
 
    def _fast_skewness(self, x: np.ndarray) -> float:
        """Optimized skewness calculation"""
        mean = np.mean(x)
        std = np.std(x)
        return np.mean(((x - mean) / std) ** 3)
 
    def _fast_kurtosis(self, x: np.ndarray) -> float:
        """Optimized kurtosis calculation"""
        mean = np.mean(x)
        std = np.std(x)
        return np.mean(((x - mean) / std) ** 4) - 3

# neural-engine/src/core/performance/gpu_accelerator.py
import cupy as cp
import numpy as np
from typing import List, Dict, Any
import torch
 
class GPUSignalProcessor:
    """GPU-accelerated signal processing using CuPy and PyTorch"""
 
    def __init__(self, device: str = 'cuda:0'):
        self.device = device
        self.torch_device = torch.device(device if torch.cuda.is_available() else 'cpu')
 
    def process_batch_gpu(self, signals: np.ndarray) -> Dict[str, Any]:
        """GPU-accelerated batch processing"""
        # Transfer to GPU
        gpu_signals = cp.asarray(signals)
 
        # GPU-accelerated FFT
        gpu_fft = cp.fft.rfft(gpu_signals, axis=1)
        gpu_power = cp.abs(gpu_fft) ** 2
 
        # GPU-accelerated statistics
        results = {
            'mean_power': cp.mean(gpu_power, axis=1),
            'max_power': cp.max(gpu_power, axis=1),
            'power_bands': self._compute_power_bands_gpu(gpu_power)
        }
 
        # Transfer results back to CPU
        return {k: cp.asnumpy(v) for k, v in results.items()}
 
    def _compute_power_bands_gpu(self, power_spectrum: cp.ndarray) -> Dict[str, cp.ndarray]:
        """Compute power in frequency bands on GPU"""
        fs = 1000  # Sampling frequency
        freqs = cp.fft.rfftfreq(power_spectrum.shape[1] * 2 - 1, 1/fs)
 
        bands = {
            'delta': (0.5, 4),
            'theta': (4, 8),
            'alpha': (8, 12),
            'beta': (12, 30),
            'gamma': (30, 100)
        }
 
        band_powers = {}
        for band_name, (low, high) in bands.items():
            mask = (freqs >= low) & (freqs < high)
            band_powers[band_name] = cp.mean(power_spectrum[:, mask], axis=1)
 
        return band_powers
 
    def process_with_pytorch(self, signals: np.ndarray, model: torch.nn.Module) -> torch.Tensor:
        """Process signals through PyTorch model on GPU"""
        # Convert to PyTorch tensor and move to GPU
        tensor_signals = torch.from_numpy(signals).float().to(self.torch_device)
 
        # Batch processing with optimal size
        batch_size = 256
        results = []
 
        with torch.no_grad():
            for i in range(0, len(tensor_signals), batch_size):
                batch = tensor_signals[i:i + batch_size]
                output = model(batch)
                results.append(output)
 
        return torch.cat(results, dim=0)

# neural-engine/src/infrastructure/database/connection_pool.py
from sqlalchemy import create_engine, pool
from sqlalchemy.orm import sessionmaker
import asyncpg
from typing import Optional
 
class DatabaseConnectionPool:
    """Optimized connection pooling for Cloud SQL"""
 
    def __init__(
        self,
        connection_string: str,
        pool_size: int = 20,
        max_overflow: int = 10,
        pool_timeout: int = 30
    ):
        # Create engine with optimized pool settings
        self.engine = create_engine(
            connection_string,
            poolclass=pool.QueuePool,
            pool_size=pool_size,
            max_overflow=max_overflow,
            pool_timeout=pool_timeout,
            pool_pre_ping=True,  # Verify connections before use
            pool_recycle=3600,   # Recycle connections after 1 hour
            echo_pool=False,
            connect_args={
                "server_settings": {
                    "jit": "off",  # Disable JIT for consistent performance
                    "application_name": "neurascale_neural_engine"
                },
                "command_timeout": 60,
                "keepalives": 1,
                "keepalives_idle": 30,
                "keepalives_interval": 10,
                "keepalives_count": 5
            }
        )
 
        self.SessionLocal = sessionmaker(
            autocommit=False,
            autoflush=False,
            bind=self.engine
        )
 
        # Async connection pool
        self.async_pool: Optional[asyncpg.Pool] = None
 
    async def init_async_pool(self):
        """Initialize async connection pool"""
        self.async_pool = await asyncpg.create_pool(
            self.connection_string,
            min_size=10,
            max_size=20,
            max_queries=50000,
            max_inactive_connection_lifetime=300,
            command_timeout=60
        )
 
    async def execute_batch_async(self, queries: List[Tuple[str, List[Any]]]):
        """Execute multiple queries efficiently"""
        async with self.async_pool.acquire() as conn:
            async with conn.transaction():
                for query, params in queries:
                    await conn.execute(query, *params)

# neural-engine/src/infrastructure/database/query_optimizer.py
from typing import List, Dict, Any
import sqlalchemy as sa
from sqlalchemy.orm import Session
from sqlalchemy import text
 
class QueryOptimizer:
    """Query optimization strategies for Cloud SQL"""
 
    def __init__(self, session: Session):
        self.session = session
 
    def bulk_insert_optimized(
        self,
        table_name: str,
        records: List[Dict[str, Any]],
        batch_size: int = 1000
    ):
        """Optimized bulk insert using COPY"""
        # Use PostgreSQL COPY for maximum performance
        for i in range(0, len(records), batch_size):
            batch = records[i:i + batch_size]
 
            # Build values clause
            values = []
            for record in batch:
                values.append(f"({','.join(str(v) for v in record.values())})")
 
            # Execute bulk insert
            query = text(f"""
                INSERT INTO {table_name} ({','.join(record.keys())})
                VALUES {','.join(values)}
                ON CONFLICT DO NOTHING
            """)
 
            self.session.execute(query)
 
        self.session.commit()
 
    def optimize_session_query(self, patient_id: str, start_time: datetime) -> Query:
        """Optimized query for session data with proper indexing"""
        return self.session.query(Session).options(
            # Eager load related data to avoid N+1 queries
            sa.orm.joinedload(Session.metrics),
            sa.orm.joinedload(Session.annotations)
        ).filter(
            # Use index-friendly filters
            Session.patient_id == patient_id,
            Session.created_at >= start_time
        ).execution_options(
            # Enable query result caching
            compiled_cache={}
        )
 
    def create_materialized_view(self, view_name: str, query: str):
        """Create materialized view for complex aggregations"""
        self.session.execute(text(f"""
            CREATE MATERIALIZED VIEW IF NOT EXISTS {view_name} AS
            {query}
            WITH DATA
        """))
 
        # Create indexes on materialized view
        self.session.execute(text(f"""
            CREATE INDEX IF NOT EXISTS idx_{view_name}_patient_id
            ON {view_name} (patient_id)
        """))

-- neural-engine/terraform/modules/database/sql/indexes.sql
 
-- Session table indexes
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_sessions_patient_id_created
ON sessions (patient_id, created_at DESC);
 
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_sessions_device_type
ON sessions (device_type)
WHERE status = 'active';
 
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_sessions_created_at_brin
ON sessions USING BRIN (created_at);
 
-- Metrics table indexes (partitioned)
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_metrics_session_id_timestamp
ON metrics (session_id, timestamp DESC);
 
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_metrics_type_timestamp
ON metrics (metric_type, timestamp DESC)
WHERE metric_type IN ('power_spectrum', 'coherence', 'phase');
 
-- Partial indexes for common queries
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_sessions_active_recent
ON sessions (created_at DESC)
WHERE status = 'active' AND created_at > NOW() - INTERVAL '7 days';
 
-- JSON indexes for metadata
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_sessions_metadata
ON sessions USING GIN (metadata jsonb_path_ops);
 
-- Text search indexes
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_annotations_search
ON annotations USING GIN (to_tsvector('english', text));
 
-- Composite indexes for analytics queries
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_analytics_composite
ON sessions (patient_id, device_type, created_at DESC)
INCLUDE (duration_seconds, quality_score);