Performance tuning

# Example: fetch_company_filings.py (line 11)
MAX_THREADS = 20  # Default

# Increase to 40 for 2x speedup (requires testing for rate limits)
MAX_THREADS = 40

import multiprocessing

# Conservative: 2x CPU cores
MAX_THREADS = multiprocessing.cpu_count() * 2

# Aggressive: 4x CPU cores (for I/O-bound tasks)
MAX_THREADS = multiprocessing.cpu_count() * 4

# Dynamic based on dataset size
total_stocks = len(stock_list)
MAX_THREADS = min(total_stocks // 10, 50)  # Cap at 50

# From fetch_fundamental_data.py
BATCH_SIZE = 100  # Fetch 100 ISINs per API call

# Increase batch size for faster execution (if API supports it)
BATCH_SIZE = 200  # 2x fewer API calls

def get_optimal_batch_size(total_items, max_threads):
    """Calculate batch size to minimize API calls while maintaining parallelism."""
    if total_items < 500:
        return 50
    elif total_items < 2000:
        return 100
    else:
        return 200

BATCH_SIZE = get_optimal_batch_size(len(stock_list), MAX_THREADS)

# Fast endpoints (indicators, announcements)
response = requests.post(url, json=payload, headers=headers, timeout=10)

# Slow endpoints (fundamental data, OHLCV)
response = requests.post(url, json=payload, headers=headers, timeout=30)

# Fast network: reduce to 5 seconds
timeout=5

# Slow/unreliable network: increase to 60 seconds
timeout=60

# run_full_pipeline.py (line 117)
result = subprocess.run(
    [sys.executable, script_path],
    cwd=BASE_DIR,
    text=True,
    timeout=1800  # 30 minutes
)

# For fetch_all_ohlcv.py on first run
if script_name == "fetch_all_ohlcv.py":
    timeout = 3600  # 1 hour for first-time full history
else:
    timeout = 1800

# fetch_company_filings.py (line 12)
FORCE_UPDATE = True  # Always re-fetch

# Change to False for incremental updates
FORCE_UPDATE = False  # Skip existing files

# Pseudocode from fetch_all_ohlcv.py
if csv_exists:
    last_date = read_last_date_from_csv()
    fetch_from = last_date + 1 day
else:
    fetch_from = 1976-10-31  # Full history

# run_full_pipeline.py (line 71)
CLEANUP_INTERMEDIATE = False  # Keep all files

# Re-run only enrichment without re-fetching base data
python3 advanced_metrics_processor.py
python3 add_corporate_events.py

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

# Create session with connection pooling
session = requests.Session()
adapter = HTTPAdapter(
    pool_connections=20,  # Number of connection pools
    pool_maxsize=50,      # Connections per pool
    max_retries=Retry(
        total=3,
        backoff_factor=0.5,
        status_forcelist=[500, 502, 503, 504]
    )
)
session.mount('http://', adapter)
session.mount('https://', adapter)

# Use session instead of requests.post()
response = session.post(url, json=payload, headers=headers, timeout=10)

import socket
import functools

# Cache DNS lookups
@functools.lru_cache(maxsize=128)
def cached_getaddrinfo(*args, **kwargs):
    return socket.getaddrinfo(*args, **kwargs)

socket.getaddrinfo = cached_getaddrinfo

import gzip

# Instead of writing JSON then compressing
with open('data.json', 'w') as f:
    json.dump(data, f)
# Then: gzip data.json

# Write compressed directly
with gzip.open('data.json.gz', 'wt') as f:
    json.dump(data, f)

import ijson

# Instead of loading entire file
with open('fundamental_data.json', 'r') as f:
    data = json.load(f)  # 35 MB into memory

# Stream process
with open('fundamental_data.json', 'rb') as f:
    for item in ijson.items(f, 'item'):
        process(item)  # Process one at a time

import gc

# Disable GC during bulk processing
gc.disable()

# Process 2,775 stocks
for stock in stock_list:
    process(stock)

# Re-enable and force collection
gc.enable()
gc.collect()

# fetch_dhan_data.py payload
"params": [
    {"field": "OgInst", "op": "", "val": "ES"},
    {"field": "Exch", "op": "", "val": "NSE"},
    {"field": "Mcap", "op": "GT", "val": "5000"},  # > ₹5,000 Cr.
    {"field": "volume", "op": "GT", "val": "500000"}  # > 500K volume
]

# run_full_pipeline.py (line 67)
FETCH_OPTIONAL = False  # Skip indices, ETFs

import subprocess
import concurrent.futures

phase2_scripts = [
    "fetch_company_filings.py",
    "fetch_new_announcements.py",
    "fetch_market_news.py",
    "fetch_corporate_actions.py",
]

def run_script(script):
    return subprocess.run(["python3", script], cwd=BASE_DIR)

with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
    futures = [executor.submit(run_script, s) for s in phase2_scripts]
    concurrent.futures.wait(futures)

# Time individual scripts
time python3 fetch_company_filings.py

# Profile Python code
python3 -m cProfile -o profile.stats fetch_company_filings.py
python3 -m pstats profile.stats
# (pstats) sort time
# (pstats) stats 10

# Add to run_full_pipeline.py
import time

phase_times = {}

start = time.time()
# ... run Phase 1 ...
phase_times['Phase 1'] = time.time() - start

start = time.time()
# ... run Phase 2 ...
phase_times['Phase 2'] = time.time() - start

# Print breakdown
for phase, duration in phase_times.items():
    print(f"{phase}: {duration:.1f}s ({duration/60:.1f} min)")