Building APIs to Track Forge Global Secondary Market Yields: A Developer's Guide to Private Market Data

# yield_calculator.py
import pandas as pd
from datetime import datetime, timedelta
import numpy as np

class PrivateMarketYieldCalculator:
    def __init__(self):
        self.risk_free_rate = 0.045  # Current 10-year Treasury
        
    def calculate_annualized_return(self, entry_price, current_value, entry_date):
        """
        Calculate annualized return for private market positions
        Accounts for illiquidity premium and time horizon
        """
        days_held = (datetime.now() - entry_date).days
        years_held = days_held / 365.25
        
        if years_held == 0:
            return 0
            
        total_return = (current_value / entry_price) - 1
        annualized_return = (1 + total_return) ** (1/years_held) - 1
        
        return annualized_return
    
    def calculate_risk_adjusted_yield(self, returns_list, benchmark_return=None):
        """
        Calculate Sharpe ratio for private market positions
        Uses estimated volatility since daily pricing unavailable
        """
        if not benchmark_return:
            benchmark_return = self.risk_free_rate
            
        avg_return = np.mean(returns_list)
        return_std = np.std(returns_list) * 1.5  # Illiquidity premium multiplier
        
        if return_std == 0:
            return 0
            
        sharpe_ratio = (avg_return - benchmark_return) / return_std
        return sharpe_ratio

# Example usage
calculator = PrivateMarketYieldCalculator()

# Sample position data
positions = [
    {
        'company': 'SpaceY Corp',
        'entry_price': 15.50,
        'current_value': 23.80,
        'entry_date': datetime(2023, 3, 15)
    },
    {
        'company': 'AI-Startup-9000',
        'entry_price': 8.25,
        'current_value': 12.10,
        'entry_date': datetime(2022, 11, 22)
    }
]

for position in positions:
    annual_return = calculator.calculate_annualized_return(
        position['entry_price'],
        position['current_value'], 
        position['entry_date']
    )
    print(f"{position['company']}: {annual_return:.2%} annualized return")

# forge_data_scraper.py
import requests
import pandas as pd
from bs4 import BeautifulSoup
import time
import json

class SecondaryMarketDataPipeline:
    def __init__(self, api_key=None):
        self.session = requests.Session()
        self.api_key = api_key
        self.base_headers = {
            'User-Agent': 'FinTech-Analytics-Bot/1.0',
            'Accept': 'application/json'
        }
        
    def fetch_company_data(self, company_symbols):
        """
        Fetch secondary market data for specified companies
        Returns pricing trends and volume data
        """
        market_data = []
        
        for symbol in company_symbols:
            try:
                # Simulated API endpoint (replace with actual data source)
                endpoint = f"https://api.secondary-markets.com/v1/companies/{symbol}"
                
                response = self.session.get(
                    endpoint,
                    headers=self.base_headers,
                    timeout=10
                )
                
                if response.status_code == 200:
                    data = response.json()
                    market_data.append({
                        'symbol': symbol,
                        'last_price': data.get('last_price'),
                        'volume_30d': data.get('volume_30d'),
                        'market_cap_estimate': data.get('market_cap_estimate'),
                        'last_funding_round': data.get('last_funding_round'),
                        'timestamp': pd.Timestamp.now()
                    })
                
                # Rate limiting - be respectful
                time.sleep(0.5)
                
            except Exception as e:
                print(f"Error fetching data for {symbol}: {e}")
                continue
        
        return pd.DataFrame(market_data)
    
    def calculate_yield_metrics(self, df):
        """
        Calculate yield metrics from market data
        """
        df['price_to_last_round'] = df['last_price'] / df['last_funding_round']
        df['volume_yield_indicator'] = df['volume_30d'] / df['market_cap_estimate']
        
        # Liquidity scoring (higher = more liquid)
        df['liquidity_score'] = (
            df['volume_30d'].rank(pct=True) * 0.6 +
            df['volume_yield_indicator'].rank(pct=True) * 0.4
        )
        
        return df

# Usage example
pipeline = SecondaryMarketDataPipeline()

# Sample company symbols for secondary market tracking
companies = ['SPACEY', 'AISTARTUP', 'FINTECH-UNICORN', 'CRYPTO-THING']

market_df = pipeline.fetch_company_data(companies)
yield_metrics = pipeline.calculate_yield_metrics(market_df)

print(yield_metrics[['symbol', 'last_price', 'liquidity_score']].head())

// yield-monitor.js
class RealTimeYieldMonitor {
  constructor(apiEndpoint, updateInterval = 60000) {
    this.apiEndpoint = apiEndpoint;
    this.updateInterval = updateInterval;
    this.positions = new Map();
    this.yieldHistory = [];
  }

  async addPosition(symbol, entryPrice, quantity, entryDate) {
    // Add position to monitoring list
    this.positions.set(symbol, {
      entryPrice: entryPrice,
      quantity: quantity,
      entryDate: new Date(entryDate),
      currentValue: entryPrice // Initialize with entry price
    });
    
    console.log(`Added ${symbol} to yield monitoring`);
  }

  async fetchCurrentPrices() {
    // Fetch latest secondary market prices
    const symbols = Array.from(this.positions.keys());
    
    try {
      const response = await fetch(`${this.apiEndpoint}/batch-prices`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ symbols: symbols })
      });
      
      const priceData = await response.json();
      
      // Update current values
      priceData.forEach(item => {
        if (this.positions.has(item.symbol)) {
          const position = this.positions.get(item.symbol);
          position.currentValue = item.price;
          position.lastUpdated = new Date();
        }
      });
      
    } catch (error) {
      console.error('Error fetching prices:', error);
    }
  }

  calculatePortfolioYield() {
    let totalCost = 0;
    let totalValue = 0;
    const yields = [];

    this.positions.forEach((position, symbol) => {
      const positionCost = position.entryPrice * position.quantity;
      const positionValue = position.currentValue * position.quantity;
      
      totalCost += positionCost;
      totalValue += positionValue;
      
      // Calculate individual position yield
      const positionYield = (positionValue - positionCost) / positionCost;
      yields.push({
        symbol: symbol,
        yield: positionYield,
        value: positionValue,
        cost: positionCost
      });
    });

    const portfolioYield = (totalValue - totalCost) / totalCost;
    
    return {
      portfolioYield: portfolioYield,
      totalValue: totalValue,
      totalCost: totalCost,
      positionYields: yields,
      timestamp: new Date()
    };
  }

  startMonitoring() {
    console.log('Starting real-time yield monitoring...');
    
    setInterval(async () => {
      await this.fetchCurrentPrices();
      const yieldData = this.calculatePortfolioYield();
      
      this.yieldHistory.push(yieldData);
      
      // Keep only last 100 data points
      if (this.yieldHistory.length > 100) {
        this.yieldHistory.shift();
      }
      
      console.log(`Portfolio Yield: ${(yieldData.portfolioYield * 100).toFixed(2)}%`);
      
    }, this.updateInterval);
  }
}

// Initialize monitoring
const monitor = new RealTimeYieldMonitor('https://api.your-secondary-data.com/v1');

// Add some positions
monitor.addPosition('SPACEY', 15.50, 100, '2023-03-15');
monitor.addPosition('AISTARTUP', 8.25, 250, '2022-11-22');

// Start monitoring
monitor.startMonitoring();

# momentum_strategy.py
import pandas as pd
import numpy as np
from scipy import stats

class SecondaryMarketMomentumStrategy:
    def __init__(self, lookback_period=90, momentum_threshold=0.15):
        self.lookback_period = lookback_period
        self.momentum_threshold = momentum_threshold
        
    def calculate_price_momentum(self, price_history):
        """
        Calculate momentum score based on price trend
        Uses linear regression slope and R-squared
        """
        if len(price_history) < self.lookback_period:
            return 0, 0
            
        recent_prices = price_history.tail(self.lookback_period)
        x = np.arange(len(recent_prices))
        y = recent_prices.values
        
        # Linear regression for trend
        slope, intercept, r_value, p_value, std_err = stats.linregress(x, y)
        
        # Normalize slope by average price
        normalized_slope = slope / np.mean(y)
        momentum_score = normalized_slope * (r_value ** 2)  # Weight by goodness of fit
        
        return momentum_score, r_value ** 2
    
    def generate_signals(self, market_data):
        """
        Generate buy/sell signals based on momentum
        """
        signals = []
        
        for symbol in market_data['symbol'].unique():
            symbol_data = market_data[market_data['symbol'] == symbol].copy()
            symbol_data = symbol_data.sort_values('timestamp')
            
            momentum, r_squared = self.calculate_price_momentum(symbol_data['last_price'])
            
            # Generate signal
            if momentum > self.momentum_threshold and r_squared > 0.6:
                signal = 'BUY'
                confidence = min(momentum * r_squared * 100, 100)
            elif momentum < -self.momentum_threshold and r_squared > 0.6:
                signal = 'SELL'
                confidence = min(abs(momentum) * r_squared * 100, 100)
            else:
                signal = 'HOLD'
                confidence = 50
            
            signals.append({
                'symbol': symbol,
                'signal': signal,
                'momentum_score': momentum,
                'confidence': confidence,
                'timestamp': pd.Timestamp.now()
            })
        
        return pd.DataFrame(signals)

# Backtesting framework
class StrategyBacktester:
    def __init__(self, initial_capital=100000):
        self.initial_capital = initial_capital
        self.current_capital = initial_capital
        self.positions = {}
        self.transaction_history = []
        
    def execute_signal(self, symbol, signal, price, confidence):
        """
        Execute trading signal with position sizing based on confidence
        """
        position_size = (confidence / 100) * 0.1  # Max 10% per position
        
        if signal == 'BUY' and symbol not in self.positions:
            investment_amount = self.current_capital * position_size
            shares = investment_amount / price
            
            self.positions[symbol] = {
                'shares': shares,
                'entry_price': price,
                'entry_date': pd.Timestamp.now()
            }
            
            self.current_capital -= investment_amount
            
            self.transaction_history.append({
                'symbol': symbol,
                'action': 'BUY',
                'shares': shares,
                'price': price,
                'timestamp': pd.Timestamp.now()
            })
            
        elif signal == 'SELL' and symbol in self.positions:
            position = self.positions[symbol]
            sale_proceeds = position['shares'] * price
            
            self.current_capital += sale_proceeds
            
            self.transaction_history.append({
                'symbol': symbol,
                'action': 'SELL',
                'shares': position['shares'],
                'price': price,
                'timestamp': pd.Timestamp.now()
            })
            
            del self.positions[symbol]
    
    def calculate_performance(self):
        """
        Calculate strategy performance metrics
        """
        total_value = self.current_capital
        
        # Add value of current positions (mark-to-market)
        for symbol, position in self.positions.items():
            # This would use current market price in real implementation
            total_value += position['shares'] * position['entry_price']
        
        total_return = (total_value - self.initial_capital) / self.initial_capital
        
        return {
            'total_return': total_return,
            'total_value': total_value,
            'cash_balance': self.current_capital,
            'active_positions': len(self.positions)
        }

# Example usage
strategy = SecondaryMarketMomentumStrategy()
backtester = StrategyBacktester()

# This would use real market data in production
sample_data = pd.DataFrame({
    'symbol': ['SPACEY'] * 100,
    'last_price': np.random.normal(20, 2, 100),
    'timestamp': pd.date_range('2023-01-01', periods=100, freq='D')
})

signals = strategy.generate_signals(sample_data)
print(signals)

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install system dependencies
RUN apt-get update && apt-get install -y \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# Copy requirements first for better caching
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY . .

# Create non-root user for security
RUN useradd --create-home --shell /bin/bash app \
    && chown -R app:app /app
USER app

# Health check endpoint
HEALTHCHECK --interval=30s --timeout=10s --start-period=60s \
  CMD python -c "import requests; requests.get('http://localhost:8000/health')"

# Run the application
CMD ["gunicorn", "--bind", "0.0.0.0:8000", "--workers", "4", "main:app"]

# docker-compose.yml
version: '3.8'

services:
  yield-tracker:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/yields
      - REDIS_URL=redis://cache:6379
    depends_on:
      - db
      - cache
    restart: unless-stopped

  db:
    image: postgres:15
    environment:
      POSTGRES_DB: yields
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
    volumes:
      - postgres_data:/var/lib/postgresql/data
    ports:
      - "5432:5432"

  cache:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  prometheus:
    image: prom/prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml

volumes:
  postgres_data:

# monitoring.py
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time
import logging

# Metrics collection
api_requests_total = Counter('api_requests_total', 'Total API requests', ['method', 'endpoint'])
yield_calculation_duration = Histogram('yield_calculation_seconds', 'Time spent calculating yields')
active_positions_gauge = Gauge('active_positions_total', 'Number of active positions')
portfolio_value_gauge = Gauge('portfolio_value_dollars', 'Total portfolio value in USD')

class PerformanceMonitor:
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        
    def track_api_request(self, method, endpoint):
        """Decorator to track API request metrics"""
        def decorator(func):
            def wrapper(*args, **kwargs):
                api_requests_total.labels(method=method, endpoint=endpoint).inc()
                return func(*args, **kwargs)
            return wrapper
        return decorator
    
    def track_yield_calculation(self, func):
        """Decorator to track yield calculation performance"""
        def wrapper(*args, **kwargs):
            with yield_calculation_duration.time():
                return func(*args, **kwargs)
        return wrapper
    
    def update_portfolio_metrics(self, positions_count, total_value):
        """Update portfolio-related metrics"""
        active_positions_gauge.set(positions_count)
        portfolio_value_gauge.set(total_value)
    
    def check_yield_thresholds(self, portfolio_yield, threshold=0.5):
        """Alert on significant yield changes"""
        if abs(portfolio_yield) > threshold:
            self.logger.warning(f"Significant yield change detected: {portfolio_yield:.2%}")
            # In production, trigger alert to Slack/email/PagerDuty

# Start Prometheus metrics server
start_http_server(8001)
monitor = PerformanceMonitor()

# test_yield_strategies.py
import unittest
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class TestYieldCalculations(unittest.TestCase):
    def setUp(self):
        self.calculator = PrivateMarketYieldCalculator()
        
    def test_annualized_return_calculation(self):
        """Test annualized return calculation accuracy"""
        # 100% return over 2 years should be ~41.42% annualized
        entry_date = datetime.now() - timedelta(days=730)
        result = self.calculator.calculate_annualized_return(
            entry_price=100,
            current_value=200,
            entry_date=entry_date
        )
        self.assertAlmostEqual(result, 0.4142, places=3)
    
    def test_risk_adjusted_yield(self):
        """Test Sharpe ratio calculation"""
        returns = [0.15, 0.25, 0.10, 0.30, 0.05]
        sharpe = self.calculator.calculate_risk_adjusted_yield(returns)
        self.assertGreater(sharpe, 0)
        
    def test_momentum_strategy_signals(self):
        """Test momentum strategy signal generation"""
        strategy = SecondaryMarketMomentumStrategy()
        
        # Create uptrend data
        dates = pd.date_range('2023-01-01', periods=100, freq='D')
        uptrend_prices = np.linspace(10, 20, 100) + np.random.normal(0, 0.5, 100)
        
        test_data = pd.DataFrame({
            'symbol': ['TEST'] * 100,
            'last_price': uptrend_prices,
            'timestamp': dates
        })
        
        signals = strategy.generate_signals(test_data)
        
        # Should generate BUY signal for strong uptrend
        self.assertEqual(signals.iloc[0]['signal'], 'BUY')
        self.assertGreater(signals.iloc[0]['confidence'], 70)

if __name__ == '__main__':
    unittest.main()