Development Guide

Complete guide for developers working with Market Intelligence.

Project Architecture

Directory Structure

market-intelligence/
├── main.py                  # CLI entry point (argparse)
├── config.py               # Configuration and env loading
├── requirements.txt        # Python dependencies
├── .env.example            # Template for API keys
│
├── agents/                 # High-level orchestration modules
│   ├── market_agent.py     # Real-time market data fetching
│   ├── news_agent.py       # News aggregation & sentiment
│   ├── fed_agent.py        # Federal Reserve & macro data
│   └── research_agent.py   # Comprehensive ticker analysis
│
├── tools/                  # Low-level utility modules
│   ├── data_sources.py     # API wrappers and data fetching
│   ├── technical.py        # Technical analysis calculations
│   ├── sentiment.py        # News sentiment scoring
│   └── formatter.py        # Terminal output formatting (Rich)
│
├── references/             # Documentation
│   ├── api-sources.md      # API integration guide
│   ├── indicator-guide.md  # Technical indicator reference
│   └── fed-calendar.md     # FOMC and macro calendar
│
├── scripts/                # Standalone utility scripts
│   ├── fetch_market.sh     # Quick shell wrapper
│   └── daily_brief.py      # Scheduled daily summary
│
├── SKILL.md                # Copilot skill definition
├── README.md               # User documentation
└── DEVELOPMENT.md          # This file

Design Patterns

1. Separation of Concerns:

   • agents/: Business logic (what to fetch)
   • tools/: Technical logic (how to fetch/calculate)
   • config/: Settings and constants
   • main.py: User interface

2. Data Flow:

   main.py (CLI)
      ↓
   agents/ (orchestration)
      ↓
   tools/ (operations)
      ↓
   API/Cache/Calculation
   

3. Error Handling:

   • Custom DataSourceError for API failures
   • Graceful fallback to cache/demo data
   • Logging via _log() function (verbose mode)

4. Caching:

   • TTLCache from cachetools (60-300 second TTL)
   • Keys: "source:params"
   • Automatic cleanup

---

API Integration Pattern

Adding a New API

1. Add to data_sources.py:

def get_new_data(symbol: str) -> Dict[str, Any]:
    """
    Fetch data from new API.
    
    Args:
        symbol: Data symbol
    
    Returns:
        Dictionary with data
    """
    cache_key = _cache_key("new_source", {"symbol": symbol})
    cached = _get_cached(cache_key)
    if cached:
        return cached
    
    try:
        # API call here
        result = {...}
        _set_cached(cache_key, result)
        return result
    except Exception as e:
        raise DataSourceError(f"API failed: {str(e)}")

2. Add config to config.py:

NEW_API_KEY = os.getenv("NEW_API_KEY", "")
NEW_API_URL = "https://api.example.com"

3. Add to .env.example:

NEW_API_KEY=your_key_here

4. Use in agent:

from tools.data_sources import get_new_data

data = get_new_data(symbol)

---

Adding New Commands

1. Define Handler in main.py

def cmd_mycommand(args):
    """Handle mycommand command."""
    print_info("Processing...")
    
    # Your logic here
    result = do_something(args)
    
    if args.output == "table":
        console.print(result)
    elif args.output == "json":
        console.print(json.dumps(result))

2. Add Parser in main()

mycommand_parser = subparsers.add_parser("mycommand", help="My command")
mycommand_parser.add_argument("--option", help="Option description")
mycommand_parser.add_argument("--output", choices=["table", "json"], default="table")

3. Add routing in main()

elif args.command == "mycommand":
    cmd_mycommand(args)

---

Technical Indicators

Adding New Indicators

In tools/technical.py:

def calculate_my_indicator(prices: pd.Series, period: int = 14) -> pd.Series:
    """
    Calculate custom indicator.
    
    Args:
        prices: Series of prices
        period: Calculation period
    
    Returns:
        Series of indicator values
    
    Formula:
        My Indicator = (Price × Factor) / Divisor
    """
    try:
        result = (prices * some_factor) / some_divisor
        return result
    except Exception as e:
        _log(f"Calculation failed: {str(e)}")
        return pd.Series(dtype=float)

Using in Research Reports

from tools.technical import calculate_my_indicator

def generate_research_report(ticker):
    hist = get_historical_data(ticker)
    prices = hist["Close"]
    
    my_ind = calculate_my_indicator(prices)
    
    report["sections"]["my_indicator"] = {
        "value": my_ind.iloc[-1],
        "signal": interpret_indicator(my_ind.iloc[-1]),
    }

---

News Sentiment Analysis

Extending Sentiment Scoring

In tools/sentiment.py:

# Add keywords
CUSTOM_KEYWORDS = {"boom", "bust", "turnaround"}

# Or use custom scoring function
def custom_sentiment_score(headline: str) -> Tuple[float, str]:
    """Custom sentiment logic."""
    # Your implementation
    return (polarity, label)

Integration

from tools.sentiment import analyze_headline_sentiment

result = analyze_headline_sentiment("AAPL beats earnings expectations")
# Returns: {"sentiment": "Bullish", "confidence": 0.87, ...}

---

Testing & Demo Mode

Demo Mode Flag

Set in config or CLI:

python main.py --demo market --ticker AAPL

Or in code:

from config import DEMO_MODE
if DEMO_MODE:
    return get_demo_data()

Demo Data Functions

All agents should have get_demo_*() functions:

def get_demo_stock_data() -> Dict[str, Any]:
    return {
        "ticker": "DEMO",
        "price": 150.00,
        ...
    }

Testing Workflow

1. Start with --demo to verify output format
2. Add single API key, test that source
3. Gradually enable all APIs
4. Test with real tickers
5. Verify error handling (invalid ticker, API down)

---

Performance Optimization

Caching Strategy

Current: 60-300 second TTL

Recommendations:

• Short-term cache (60s): Real-time data
• Medium cache (300s): Historical/technical
• Long-term cache (24h): Fundamentals

Rate Limiting

Built-in strategies:

1. Respect API rate limits
2. Batch requests when possible
3. Use cache before re-requesting
4. Implement backoff/retry

Async/Await (Future)

Currently synchronous. For async support:

import asyncio
from aiohttp import ClientSession

async def fetch_all(tickers):
    tasks = [fetch_one(ticker) for ticker in tickers]
    return await asyncio.gather(*tasks)

---

Logging & Debugging

Enabling Verbose Mode

python main.py --verbose market --ticker AAPL

Or:

VERBOSE=true

_log() Function Pattern

Used throughout codebase:

from config import VERBOSE

def _log(message: str):
    if VERBOSE:
        print(f"[DEBUG] {message}")

Adding Debug Output

_log(f"Fetching data for {ticker}")
_log(f"Cache hit: {key}")
_log(f"API response: {response}")

---

Documentation Standards

Docstring Format

def function_name(param1: str, param2: int) -> Dict[str, Any]:
    """
    Short description (1 line).
    
    Longer description if needed, explaining purpose and behavior.
    Can include implementation notes.
    
    Args:
        param1: Description of param1
        param2: Description of param2
    
    Returns:
        Description of return value
    
    Raises:
        CustomError: When error condition occurs
    
    Example:
        >>> result = function_name("value", 10)
        >>> result["key"]
        "expected"
    """

Type Hints

Always use type hints:

def process_data(
    data: List[Dict[str, Any]],
    filter_type: Optional[str] = None
) -> Dict[str, Any]:
    pass

---

Common Debugging Issues

Issue: "ModuleNotFoundError"

Solution: Ensure you're in project directory, and did pip install -r requirements.txt

Issue: "API key not found"

Solution: Check .env file exists and has correct format

Issue: "Rate limit exceeded"

Solution: Use --demo flag or wait 60 seconds (cache TTL)

Issue: "Ticker not found"

Solution: Verify ticker is valid (case-sensitive for crypto)

Issue: "No data returned"

Solution:

1. Check if market is closed
2. Verify API is responding
3. Enable --verbose to see API calls
4. Check network connectivity

---

Contributing Guidelines

Code Style

• Use 4 spaces for indentation
• Line length: 80 characters (docstrings, 100)
• PEP 8 compliant
• Type hints on all functions
• Docstrings on all public functions

Before Committing

# Run linter
pylint *.py tools/*.py agents/*.py

# Run formatter
black --line-length 80 *.py tools/*.py agents/*.py

# Test basic functionality
python main.py --demo snapshot
python main.py --demo research --ticker AAPL

Adding Features

1. Create feature branch: git checkout -b feature/my-feature
2. Add tests/demos
3. Document in docstrings
4. Update README if user-facing
5. Update this DEVELOPMENT.md if architectural
6. Create pull request

---

Future Enhancements

Planned Features

• [ ] Async/await for parallel requests
• [ ] Real-time WebSocket connections
• [ ] Portfolio tracking and alerts
• [ ] Risk calculation and optimization
• [ ] Machine learning predictions
• [ ] Database storage (SQLite)
• [ ] Web dashboard (Streamlit)
• [ ] Mobile app (React Native)

Known Limitations

• No real-time data (15-20 min delay from yfinance)
• No options pricing
• Limited to free API tiers
• Single-threaded execution
• No user authentication
• No data persistence

---

Resources

• SKILL.md: Copilot skill definition
• README.md: User-facing documentation
• references/api-sources.md: API integration details
• references/indicator-guide.md: Technical analysis reference
• references/fed-calendar.md: FOMC and macro calendar

---

Support & Questions

For issues:

1. Check this file and SKILL.md
2. Review error messages with --verbose
3. Check references/ folder
4. Inspect relevant source files
5. Try --demo mode to isolate issue