PRODUCTION_CHECKLIST.md

Before Production: What to Add

This example is bare minimum for demonstration. Before using in production, add:

Logging (Required)

Current: Simple print statements Add: Proper logging to file

import logging

# Add to main():
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
handler = logging.FileHandler('market_data.log')
logger.addHandler(handler)
logger.info(f"Retrieved {len(data)} rows")

Error Handling (Required)

Current: Basic try/except Add: Specific exception handling, retries, and monitoring

try:
    data = ld.get_data(...)
except ConnectionError:
    logger.error("Connection failed")
    # Retry logic here
except Exception as e:
    logger.error(f"Error: {e}", exc_info=True)
    # Send alert/email

Data Validation (Required)

Current: No validation Add: Check data quality before saving

if data.empty:
    logger.error("No data returned")
    return 1

if len(data) < expected_count:
    logger.warning(f"Expected {expected_count}, got {len(data)}")

Configuration File (Recommended)

Current: Hardcoded values Add: Config file for instruments and fields

# config.json
{
  "instruments": ["IBM.N", "MSFT.O", "APPL.O"],
  "fields": ["TR.PriceClose", "TR.Volume"],
  "output_dir": "C:\\market_data"
}

Session Management (Recommended)

Current: No timeout handling Add: Session initialization check and timeout

try:
    session = ld.open_session()
except Exception as e:
    logger.error("Failed to open session")
    # Ensure session closes even on failure

Monitoring & Alerts (Recommended)

Current: No notifications Add: Email/Slack alerts on failure

if exit_code != 0:
    send_alert(f"Market data job failed: {error_msg}")

Logging (Optional but Useful)

Add structured logging:

import logging.handlers

# Rotate logs by size
handler = logging.handlers.RotatingFileHandler(
    'market_data.log',
    maxBytes=10*1024*1024,  # 10MB
    backupCount=10
)

Database Storage (Optional)

Current: CSV files only Add: Database for historical tracking

# Store in database instead of (or in addition to) CSV
# Allows queries like: "Get all closes for IBM in last 30 days"

Schedule Configuration (Recommended)

Current: Hardcoded 7:00 AM Add: Make schedule configurable

# environment variables or config file
SCHEDULE_TIME = os.getenv('SCHEDULE_TIME', '07:00')
SCHEDULE_DAYS = os.getenv('SCHEDULE_DAYS', 'MON,TUE,WED,THU,FRI')

Testing (Required for Production)

Add unit tests:

def test_csv_creation():
    result = main()
    assert result == 0
    assert Path('market_data_*.csv').exists()

def test_invalid_instruments():
    result = main(['INVALID.X'])
    assert result == 1

Security (Required for Production)

Current: No credential handling Add: Secure credential management

# Use environment variables or secure vaults
# NEVER commit credentials to version control
API_KEY = os.getenv('LSEG_API_KEY')

Documentation (Recommended)

Add:

• Runbook for manual execution
• Troubleshooting guide for ops team
• SLA/performance expectations
• Disaster recovery procedures

Batch File Improvements

Current: Simple run Add:

REM Better error handling
if %ERRORLEVEL% neq 0 (
    REM Send alert
    REM Retry
)

REM Start LSEG Workspace if needed
REM Wait for it to be ready
REM Capture output for audit

REM Check for previous failure
REM Only run if previous run succeeded

Monitoring Dashboard

Track:

• Script execution time
• Number of rows returned
• Error rates
• File creation timestamps
• CSV file sizes

Deployment Checklist

• Logging configured
• Error handling for all edge cases
• Data validation added
• Configuration file created
• Credentials in environment variables
• Unit tests pass
• Integration tests pass
• Monitoring/alerts configured
• Documentation complete
• Backup procedures in place
• Tested on target environment
• Runbook created for ops team

Minimum Production Version Example

import logging
import os
from datetime import datetime
from pathlib import Path
from lseg import data as ld

logging.basicConfig(
    filename='market_data.log',
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

def main():
    logger = logging.getLogger(__name__)
    try:
        session = ld.open_session()
        data = ld.get_data(['IBM.N', 'MSFT.O'], ['TR.PriceClose'])
        
        if data.empty:
            logger.error("No data returned")
            return 1
        
        csv_file = f"market_data_{datetime.now().strftime('%Y%m%d')}.csv"
        data.to_csv(csv_file)
        logger.info(f"Saved {len(data)} rows to {csv_file}")
        
        return 0
    except Exception as e:
        logger.error(f"Error: {e}", exc_info=True)
        return 1
    finally:
        try:
            session.close()
        except:
            pass

if __name__ == '__main__':
    exit(main())

---

Start with this bare minimum example to demonstrate the concept, then upgrade piece by piece based on your requirements.