The Command pattern intuitively works like placing a takeout order 🥡 at a restaurant. It’s all about separating the person who asks for the food from the chef who makes it, using a ticket (the Command object) as the medium. So there is the client, the command/order ticket, invoker/order taker and receiver/chef.

the Command pattern means the request is not directly commanded but is decomposed into several distinct roles to achieve decoupling and flexibility. This separation is what enables powerful features like macro recording, undo/redo, and transactional logging. Why? because it turns an action (a verb) into an object (a noun). Once the action is an object, it can be treated like data: stored, manipulated, and recalled, rather than just being executed and forgotten.

from abc import ABC, abstractmethod

class OrderCommand(ABC):
    @abstractmethod
    def execute(self):
        pass

class BuyOrder(OrderCommand):
    def __init__(self, broker, symbol, qty):
        self.broker = broker
        self.symbol = symbol
        self.qty = qty

    def execute(self):
        self.broker.buy(self.symbol, self.qty)

class SellOrder(OrderCommand):
    def __init__(self, broker, symbol, qty):
        self.broker = broker
        self.symbol = symbol
        self.qty = qty

    def execute(self):
        self.broker.sell(self.symbol, self.qty)

class BrokerSystem:
    def buy(self, symbol, qty):
        print(f"Buying {qty} of {symbol}")

    def sell(self, symbol, qty):
        print(f"Selling {qty} of {symbol}")

class Trader:
    def __init__(self):
        self.commands = []

    def place_order(self, command):
        self.commands.append(command)
        command.execute()

broker = BrokerSystem()
trader = Trader()
trader.place_order(BuyOrder(broker, "AAPL", 100))
trader.place_order(SellOrder(broker, "GOOG", 50))

# without command pattern, it's messy, it only have BrokerSystem and Trader Classes
class BrokerSystem:
    def buy(self, symbol, qty):
        print(f"Buying {qty} of {symbol}")
    
    def sell(self, symbol, qty):
        print(f"Selling {qty} of {symbol}")

class Trader:
    def __init__(self, broker):
        self.broker = broker
        self.order_history = []
    
    def buy(self, symbol, qty):
        # Directly call broker method
        self.broker.buy(symbol, qty)
        self.order_history.append(("buy", symbol, qty))
    
    def sell(self, symbol, qty):
        # Directly call broker method
        self.broker.sell(symbol, qty)
        self.order_history.append(("sell", symbol, qty))
    
    def cancel_order(self, symbol):
        # Need another method for cancellation
        print(f"Cancelled order for {symbol}")
    
    def amend_order(self, symbol, new_qty):
        # Need another method for amendment
        print(f"Amended order for {symbol} to {new_qty}")

# Usage
broker = BrokerSystem()
trader = Trader(broker)
trader.buy("AAPL", 100)
trader.sell("GOOG", 50)
trader.cancel_order("AAPL")
trader.amend_order("GOOG", 75)

Composite Pattern is common in indexing workflow, as we will have individual fund and fund of funds, and the two some time need to be treated equally as an object, Composite patterns comes to play.

class PortfolioComponent:
    def value(self):
        raise NotImplementedError

class Fund(PortfolioComponent):
    def __init__(self, name, val):
        self.name = name
        self._val = val

    def value(self):
        return self._val # skip the method and just expose _val directly, you lose the ability to change implementation details later without breaking everything that depends on your class.

class Portfolio(PortfolioComponent):
    def __init__(self, name):
        self.name = name
        self.children = []

    def add(self, component):
        self.children.append(component)

    def value(self):
        return sum(c.value() for c in self.children)

# Compose hierarchy
equity_fund = Fund("Equity Fund", 1_000_000)
bond_fund = Fund("Bond Fund", 500_000)
fund_of_funds = Portfolio("Global Portfolio")
fund_of_funds.add(equity_fund)
fund_of_funds.add(bond_fund)

print(f"{fund_of_funds.name} total value = {fund_of_funds.value():,.0f}")

Template Method: in index calculation workflow, steps are consistent but details can vary. the below example codes demonstrate the scenario when only weighting varies.

from abc import ABC, abstractmethod

class IndexCalculationTemplate(ABC):
    def run(self):
        self.load_data()
        self.clean_data()
        self.apply_weights()
        self.compute_return()

    def load_data(self):
        print("Loading market data...")

    def clean_data(self):
        print("Cleaning missing data...")

    @abstractmethod
    def apply_weights(self):
        pass

    def compute_return(self):
        print("Computing index return...")

class EqualWeightedIndex(IndexCalculationTemplate):
    def apply_weights(self):
        print("Applying equal weights to all stocks.")

class MarketCapWeightedIndex(IndexCalculationTemplate):
    def apply_weights(self):
        print("Applying market-cap weights.")

EqualWeightedIndex().run()
MarketCapWeightedIndex().run()

State Pattern is Perfect for Version Management:

from abc import ABC, abstractmethod
from datetime import datetime

# State Interface
class IndexState(ABC):
    @abstractmethod
    def review(self, index):
        pass
    
    @abstractmethod
    def approve(self, index):
        pass
    
    @abstractmethod
    def publish(self, index):
        pass
    
    @abstractmethod
    def get_status(self):
        pass

# Concrete States
class DraftState(IndexState):
    """Index is being created/edited"""
    def review(self, index):
        print("✓ Index moved to Review state")
        index.state = ReviewState()
    
    def approve(self, index):
        print("✗ Cannot approve. Must review first.")
    
    def publish(self, index):
        print("✗ Cannot publish from Draft. Must review and approve.")
    
    def get_status(self):
        return "DRAFT - Editing in progress"

class ReviewState(IndexState):
    """Index is under review"""
    def review(self, index):
        print("✗ Already in review state")
    
    def approve(self, index):
        print("✓ Index approved. Moving to Approved state")
        index.state = ApprovedState()
    
    def publish(self, index):
        print("✗ Cannot publish. Must be approved first.")
    
    def get_status(self):
        return "REVIEW - Awaiting approval"

class ApprovedState(IndexState):
    """Index is approved and ready to publish"""
    def review(self, index):
        print("✗ Cannot review approved index. Move back to Draft if changes needed.")
    
    def approve(self, index):
        print("✗ Already approved")
    
    def publish(self, index):
        print("✓ Index published successfully!")
        index.state = PublishedState()
        index.published_date = datetime.now()
    
    def get_status(self):
        return "APPROVED - Ready for publication"

class PublishedState(IndexState):
    """Index is live"""
    def review(self, index):
        print("✗ Cannot review published index. Create new version instead.")
    
    def approve(self, index):
        print("✗ Already published")
    
    def publish(self, index):
        print("✗ Already published")
    
    def get_status(self):
        return "PUBLISHED - Live"

class DeprecatedState(IndexState):
    """Index version is deprecated"""
    def review(self, index):
        print("✗ Cannot modify deprecated index. Create new version.")
    
    def approve(self, index):
        print("✗ Cannot approve deprecated index")
    
    def publish(self, index):
        print("✗ Cannot publish deprecated index")
    
    def get_status(self):
        return "DEPRECATED - No longer maintained"

# Index Context
class IndexVersion:
    def __init__(self, name, version, constituents):
        self.name = name
        self.version = version
        self.constituents = constituents
        self.state = DraftState()  # Initial state
        self.published_date = None
        self.created_date = datetime.now()
    
    def review(self):
        self.state.review(self)
    
    def approve(self):
        self.state.approve(self)
    
    def publish(self):
        self.state.publish(self)
    
    def deprecate(self):
        print(f"✓ Index v{self.version} deprecated")
        self.state = DeprecatedState()
    
    def show_status(self):
        return f"\n{self.name} v{self.version}:\n  Status: {self.state.get_status()}\n  Constituents: {self.constituents}"

# Usage Demo
print("=" * 60)
print("Index Version Management - State Pattern Demo")
print("=" * 60)

# Create new index version
sp500_v1 = IndexVersion("S&P 500", "1.0", ["AAPL", "MSFT", "GOOG"])
print(sp500_v1.show_status())

# Workflow: Draft → Review → Approved → Published
print("\n--- Workflow Progression ---")
sp500_v1.review()
print(sp500_v1.show_status())

sp500_v1.approve()
print(sp500_v1.show_status())

sp500_v1.publish()
print(sp500_v1.show_status())

# Try invalid transitions
print("\n--- Invalid Transitions (prevented by state) ---")
sp500_v1.review()  # ✗ Cannot review published
sp500_v1.approve()  # ✗ Cannot approve published

# Create new version when old one is deprecated
print("\n--- Version Management ---")
sp500_v2 = IndexVersion("S&P 500", "2.0", ["AAPL", "MSFT", "GOOG", "AMZN", "TSLA"])
print(sp500_v2.show_status())
sp500_v2.review()
sp500_v2.approve()
sp500_v2.publish()
print(sp500_v2.show_status())

# Deprecate old version
sp500_v1.deprecate()
print(sp500_v1.show_status())

print("\n--- Multiple Versions Active ---")
print(f"Version 1: {sp500_v1.state.get_status()}")
print(f"Version 2: {sp500_v2.state.get_status()}")

Proxy Pattern: You want to control who can access certain calculations or data, such as proprietary index formulas or expensive API calls. The proxy is like adding a hook (gate/checkpoint) before the real object.

class IndexCalculator:
    def calculate(self, name):
        print(f"Calculating {name} index...")

class SecureProxy:
    def __init__(self, real_calc, user_role):
        self.real_calc = real_calc
        self.user_role = user_role

    def calculate(self, name):
        if self.user_role == "admin":
            self.real_calc.calculate(name)
        else:
            print("Access denied: insufficient permissions")

real = IndexCalculator()
secure_calc = SecureProxy(real, user_role="analyst")
secure_calc.calculate("AI Equity Index")

A large universe of securities (say 10,000+) share common attributes like sector, country, or currency.
You don’t want to store these redundant strings in every object. Use Flyweight pattern.

Interpreter Pattern — “Rule-based Strategy Language” but sometime it could be an overkill and verbose

Below codes illustrate how a complete simple workflow of indexing can be applied:

# patterns_demo.py  (conceptual)
from abc import ABC, abstractmethod
import copy
from typing import List, Dict, Any

# ---------- Strategy: weighting algorithms ----------
class WeightStrategy(ABC):
    @abstractmethod
    def compute_weights(self, constituents: Dict[str, float]) -> Dict[str, float]:
        pass

class MarketCapWeighting(WeightStrategy):
    def compute_weights(self, constituents):
        total = sum(constituents.values())
        return {t: v/total for t, v in constituents.items()}

class EqualWeighting(WeightStrategy):
    def compute_weights(self, constituents):
        n = len(constituents)
        return {t: 1/n for t in constituents}

# ---------- Builder: construct an index ----------
class Index:
    def __init__(self, name: str):
        self.name = name
        self.constituents: Dict[str, float] = {}  # ticker -> market cap (or score)
        self.weights: Dict[str, float] = {}

    def __repr__(self):
        return f"Index({self.name}, {len(self.constituents)} constituents)"

class IndexBuilder:
    def __init__(self):
        self._index = None

    def new_index(self, name: str):
        self._index = Index(name)
        return self

    def load_universe(self, universe: Dict[str, float]):
        # universe: ticker -> market cap (just an example)
        self._index.constituents = dict(universe)
        return self

    def apply_filters(self, min_mcap=0):
        self._index.constituents = {t:m for t,m in self._index.constituents.items() if m >= min_mcap}
        return self

    def build(self):
        return self._index

# ---------- Memento: snapshotting index state ----------
class IndexMemento:
    def __init__(self, state: Dict[str, Any]):
        self._state = copy.deepcopy(state)

    def get_state(self):
        return copy.deepcopy(self._state)

class IndexHistory:
    def __init__(self):
        self._stack: List[IndexMemento] = []

    def push(self, m: IndexMemento):
        self._stack.append(m)

    def pop(self) -> IndexMemento:
        return self._stack.pop() if self._stack else None

# ---------- Observer: market data feed ----------
class MarketDataFeed:
    def __init__(self):
        self._subs = []

    def subscribe(self, handler):
        self._subs.append(handler)

    def publish(self, price_map: Dict[str, float]):
        for h in self._subs:
            h.update_prices(price_map)

# ---------- Index Service: ties patterns together ----------
class IndexService:
    def __init__(self, strategy: WeightStrategy):
        self.strategy = strategy
        self._history = IndexHistory()
        self._index: Index | None = None

    def construct_index(self, name, universe):
        builder = IndexBuilder()
        self._index = builder.new_index(name).load_universe(universe).apply_filters(min_mcap=0).build()
        return self._index

    def snapshot(self):
        if not self._index:
            return
        state = {"name": self._index.name, "constituents": self._index.constituents, "weights": self._index.weights}
        self._history.push(IndexMemento(state))

    def restore(self):
        m = self._history.pop()
        if m:
            state = m.get_state()
            self._index = Index(state["name"])
            self._index.constituents = state["constituents"]
            self._index.weights = state["weights"]

    def rebalance(self):
        if not self._index:
            raise RuntimeError("No index")
        self.snapshot()  # save before mutate
        self._index.weights = self.strategy.compute_weights(self._index.constituents)
        return self._index.weights

    # Observer handler interface
    def update_prices(self, price_map: Dict[str, float]):
        # reactive update: update market caps (example)
        if not self._index:
            return
        # Update constituents' market caps and optionally trigger rebalance
        for t, price in price_map.items():
            if t in self._index.constituents:
                # simplistic: treat 'constituents[t]' as mcap and scale it by price
                self._index.constituents[t] = price  # in real life you'd multiply by shares outstanding
        # Optionally rebalance on big change (policy)
        # self.rebalance()

# ---------- Example usage ----------
if __name__ == "__main__":
    universe = {"AAPL": 2_000, "MSFT": 1_500, "GOOG": 1_200}
    svc = IndexService(strategy=MarketCapWeighting())
    idx = svc.construct_index("TechIndex", universe)
    svc.rebalance()
    print(idx.weights)

    # subscription to market data
    feed = MarketDataFeed()
    feed.subscribe(svc)
    feed.publish({"AAPL": 2050, "GOOG": 1250})
    svc.rebalance()
    print(idx.weights)

    # undo last rebalance
    svc.restore()
    print("Restored weights:", idx.weights)