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())