Patterns in Finance and Indexing 01

First, the adapter pattern should be implemented to standardize the data feed intake. It is crucial to establish a uniform method, such as data_provider.get_price(“AAPL”), to accommodate the various API formats provided by different vendors.

class BloombergAPI:
    def getField(self, ticker, field):
        return f"Bloomberg {ticker}:{field}=185.5"

class FactSetAPI:
    def get_price(self, ticker):
        return f"FactSet {ticker}=185.5"


# Unified interface
class MarketDataProvider:
    def get_price(self, ticker):
        raise NotImplementedError


# Adapters
class BloombergAdapter(MarketDataProvider):
    def __init__(self, bloomberg_api):
        self.api = bloomberg_api

    def get_price(self, ticker):
        return self.api.getField(ticker, "PX_LAST")


class FactSetAdapter(MarketDataProvider):
    def __init__(self, factset_api):
        self.api = factset_api

    def get_price(self, ticker):
        return self.api.get_price(ticker)
# usage
bloomberg = BloombergAdapter(BloombergAPI())
factset = FactSetAdapter(FactSetAPI())

for provider in [bloomberg, factset]:
    print(provider.get_price("AAPL"))

Two, Facade Pattern to simply complex analytics APIs, only one pretty facade is exposed to the user.

class PortfolioEngine:
    def calc_nav(self): return 1000

class RiskEngine:
    def calc_var(self): return 25

class ComplianceChecker:
    def check_limits(self): return "All limits OK"


class IndexAnalyticsFacade:
    def __init__(self):
        self.portfolio = PortfolioEngine()
        self.risk = RiskEngine()
        self.compliance = ComplianceChecker()

    def get_summary_report(self):
        nav = self.portfolio.calc_nav()
        var = self.risk.calc_var()
        status = self.compliance.check_limits()
        return {
            "NAV": nav,
            "VaR": var,
            "Compliance": status
        }
analytics = IndexAnalyticsFacade()
report = analytics.get_summary_report()
print(report)

Third, Bridge Pattern to decouple index logic and data source, two hierarchy systems can be bridged together so there are much less combination complexity, i.e. decreasing complexity from nxm to n+m.

# Implementation hierarchy
class DataProvider:
    def get_price(self, ticker):
        raise NotImplementedError

class FactSetData(DataProvider):
    def get_price(self, ticker):
        return 185.5  # mock data

class BloombergData(DataProvider):
    def get_price(self, ticker):
        return 186.0


# Abstraction hierarchy
class Index:
    def __init__(self, data_provider):
        self.data_provider = data_provider

    def calculate(self, tickers):
        raise NotImplementedError


class PriceWeightedIndex(Index):
    def calculate(self, tickers):
        prices = [self.data_provider.get_price(t) for t in tickers]
        return sum(prices) / len(prices)


class MarketCapWeightedIndex(Index):
    def calculate(self, tickers):
        prices = [self.data_provider.get_price(t) for t in tickers]
        weights = [0.4, 0.3, 0.3]
        return sum(p * w for p, w in zip(prices, weights))
# usage
tickers = ["AAPL", "MSFT", "GOOG"]

index1 = PriceWeightedIndex(BloombergData())
index2 = MarketCapWeightedIndex(FactSetData())

print("Bloomberg price-weighted:", index1.calculate(tickers))
print("FactSet market-cap:", index2.calculate(tickers))

Prototype, instead of reating everything over and over again, clone an existing object and modify it, that’s the essence of prototype pattern, for example, I have S&P500 index and would like to create equal weighted or carbon-reduced variations.

import copy

class IndexPrototype:
    def clone(self):
        return copy.deepcopy(self)

class Index(IndexPrototype):
    def __init__(self, name, constituents):
        self.name = name
        self.constituents = constituents

    def __str__(self):
        return f"{self.name}: {len(self.constituents)} stocks"

# Create base index
sp500 = Index("S&P 500", ["AAPL", "MSFT", "GOOG"])

# Clone and modify
equal_weighted = sp500.clone()
equal_weighted.name = "S&P 500 Equal Weight"

print(sp500)
print(equal_weighted)

In Index rebalancing, it could be performed time-based, threshold-based or volatility-based etc., we need to encapsulate each strategy into its own class and swap them dynamically. This is Strategy pattern.

from abc import ABC, abstractmethod

class RebalanceStrategy(ABC):
    @abstractmethod
    def rebalance(self, portfolio):
        pass

class TimeBasedRebalance(RebalanceStrategy):
    def rebalance(self, portfolio):
        print("Rebalancing by time schedule...")

class ThresholdRebalance(RebalanceStrategy):
    def rebalance(self, portfolio):
        print("Rebalancing when deviation exceeds threshold...")

class Portfolio:
    def __init__(self, name, strategy: RebalanceStrategy):
        self.name = name
        self.strategy = strategy

    def execute_rebalance(self):
        self.strategy.rebalance(self)

# Use strategies
portfolio = Portfolio("AI Equity Index", TimeBasedRebalance())
portfolio.execute_rebalance()

portfolio.strategy = ThresholdRebalance()
portfolio.execute_rebalance()

Observer pattern to solve the problem when market data change, multiple systems need to react, so observer pattern helps broadcast updates efficiently.

class MarketDataFeed:
    def __init__(self):
        self.subscribers = []

    def subscribe(self, observer):
        self.subscribers.append(observer)

    def notify(self, price):
        for obs in self.subscribers:
            obs.update(price)

class IndexCalculator: # it's observer
    def update(self, price):
        print(f"Updating index value with new price {price}")

class RiskMonitor: # it's observer
    def update(self, price):
        print(f"Recomputing risk based on price {price}")

feed = MarketDataFeed()
feed.subscribe(IndexCalculator()) 
feed.subscribe(RiskMonitor())

feed.notify(4523.88)

Decorator pattern is very important, in indexing workflow, if you want to add features to your index analytics — such as tracking error, volatility, drawdown, or Sharpe ratio — without changing the base class. In the following example, BaseAnalytics is decorated by VolatilityDecorator, then further decorated by SharepeDecorator.

class BaseAnalytics:
    def analyze(self):
        return "Basic return stats"

class VolatilityDecorator:
    def __init__(self, analytics):
        self.analytics = analytics

    def analyze(self):
        return self.analytics.analyze() + " + Volatility computed"

class SharpeDecorator:
    def __init__(self, analytics):
        self.analytics = analytics

    def analyze(self):
        return self.analytics.analyze() + " + Sharpe ratio computed"

# Layer them
analytics = SharpeDecorator(VolatilityDecorator(BaseAnalytics()))
print(analytics.analyze())

Leave a comment

This site uses Akismet to reduce spam. Learn how your comment data is processed.