Structural Behavioral Patterns

Structural & Behavioral Design Patterns

Applies GoF structural and behavioral design patterns to decouple components, manage object responsibilities, and establish clean communication channels between classes. This skill makes the model select the right pattern for interface adaptation (Adapter), feature composition (Bridge, Composite, Decorator, Facade, Proxy, Flyweight), runtime behavior swapping (Strategy), request encapsulation (Command), template-based inheritance (Template Method), event-driven decoupling (Observer), and workflow orchestration (Mediator, Chain of Responsibility).

TL;DR Checklist

• Identify the structural problem first: incompatible interfaces (Adapter), tree structures (Composite), dynamic augmentation (Decorator), simplified APIs (Facade), access control (Proxy), shared data (Flyweight)
• Identify the behavioral problem first: interchangeable algorithms (Strategy), request encapsulation with undo (Command), template-based inheritance (Template Method), event pub/sub (Observer), workflow delegation (Mediator), sequential processing (Chain of Responsibility)
• Prefer composition over inheritance for all structural and behavioral patterns
• Use typing.Protocol or ABC interfaces to ensure transparent substitution of pattern participants
• Enforce Open/Closed Principle through polymorphism, not conditional logic branches
• Ensure Facade does not become a god object — delegate to domain services
• Keep Command objects immutable where possible for thread-safe undo/redo stacks
• Use Observer with weak references to prevent memory leaks from forgotten unsubscription
• Avoid over-engineering: do not apply patterns where simple function calls suffice

When to Use

Use this skill when:

• Two existing classes have incompatible interfaces and you need them to work together without modifying either class's source
• You need to add new responsibilities to objects dynamically at runtime instead of via subclassing (Decorator)
• A subsystem has a complex API and you want a simplified unified interface for common operations (Facade)
• You want to swap algorithms or strategies at runtime based on context or configuration (Strategy)
• You need to encapsulate requests as objects so you can parameterize clients with different requests, queue them, log them, or support undo/redo (Command)
• Multiple objects need to be notified when another object's state changes, and you want loose coupling between the subject and its observers (Observer)
• You have a family of related operations that share common steps but differ in specific details (Template Method)
• You are building a tree-like part-whole hierarchy where clients should treat individual objects and compositions uniformly (Composite)
• You need to control access to an object, defer initialization, or add caching/authentication layers without changing the original class (Proxy)
• You have many objects that share intrinsic state and can safely share it across instances (Flyweight)

When NOT to Use

Avoid this skill for:

• Simple functions or methods that accomplish the task directly — do not introduce a pattern for pattern's sake (see YAGNI)
• Inheritance hierarchies where subclassing cleanly covers the variation — Template Method may apply, but prefer Strategy for runtime swaps
• When you can use composition and direct method calls without needing decoupled event notification — Observer adds unnecessary indirection
• Performance-critical inner loops where the indirection overhead of a pattern matters — inline the logic directly
• Domain models that already have clear responsibility boundaries — do not force patterns onto well-designed code (see SOLID as the primary guide)

Core Workflow

1. Identify the Structural vs. Behavioral Problem — Determine whether the problem is about object composition/structure (Adapter, Bridge, Composite, Decorator, Facade, Proxy, Flyweight) or about object responsibilities and communication (Observer, Strategy, Command, Template Method, Mediator, Chain of Responsibility). Checkpoint: If the problem is "make these two interfaces work together," it is structural. If it is "change behavior at runtime" or "notify multiple objects," it is behavioral.

2. Select the Pattern Based on the Specific Problem — Map the problem to the correct pattern using the decision matrix:

   • Incompatible interface → Adapter
   • Need abstraction separated from implementation details → Bridge
   • Tree-like part-whole hierarchy → Composite
   • Add behaviors dynamically without subclassing → Decorator
   • Simplify complex subsystem API → Facade
   • Control/lazy-load/access-check an object → Proxy
   • Share data among many instances → Flyweight
   • Swap algorithms at runtime → Strategy
   • Encapsulate requests with undo/redo → Command
   • Define algorithm skeleton in a base class → Template Method
   • Decouple event publishers from subscribers → Observer
   • Centralize communication between many objects → Mediator
   • Process requests through multiple handlers sequentially → Chain of Responsibility

3. Define the Interfaces (Protocols) — Use typing.Protocol or abc.ABC to define the common interface that all concrete participants will implement. This is the critical step that ensures type safety and polymorphic substitution. Checkpoint: Every concrete class in the pattern must satisfy the protocol/ABC interface. Verify with static type checking before proceeding.

4. Implement Participants with Composition — Build each pattern participant as a separate class, connected through composition (aggregation) rather than inheritance where possible. Each class should have a single responsibility aligned with its role in the pattern. Checkpoint: No class should do more than what the GoF definition of its role requires. Apply SRP strictly.

5. Wire Participants Together — Connect the subject/client to the adaptee/implementation/strategy/command via dependency injection (constructor or setter). Ensure the client interacts only with the abstract interface, never with concrete implementations directly. Checkpoint: The client should compile and work if you swap any concrete participant for another that satisfies the same protocol.

6. Add Error Handling and Edge Cases — Implement guard clauses in pattern entry points (e.g., empty observer lists in Observer, null commands in Command). Validate state transitions where applicable (e.g., command state before undo). Checkpoint: Pattern runtime should never crash due to an empty collection, missing handler, or null reference.

Implementation Patterns / Reference Guide

Pattern 1: Adapter Pattern

Makes incompatible interfaces work together by wrapping one interface in a new interface that clients expect. The adapter translates calls to the adaptee's native interface.

Use Adapter when you must integrate a third-party library, legacy code, or an external service whose API does not match what your application expects. It is the only structural pattern that deals with interface incompatibility rather than object structure.

from __future__ import annotations
import logging
from typing import Protocol, runtime_checkable

logger = logging.getLogger(__name__)


@runtime_checkable
class PaymentProcessor(Protocol):
    """The target interface expected by the application."""

    def process_payment(self, amount: float, currency: str) -> bool: ...


@runtime_checkable
class LegacyPaymentGateway(Protocol):
    """The existing (incompatible) interface we need to adapt."""

    def send_payment_request(self, value: int, code: str) -> dict[str, str]: ...


class PaymentGatewayAdapter(PaymentProcessor):
    """Adapts LegacyPaymentGateway to the PaymentProcessor target interface.

    The adapter sits between the client (which expects PaymentProcessor)
    and the legacy system (which exposes LegacyPaymentGateway). It translates
    parameter names, units, and response formats.
    """

    def __init__(self, gateway: LegacyPaymentGateway) -> None:
        self._gateway = gateway

    def process_payment(self, amount: float, currency: str) -> bool:
        try:
            # Convert from dollars (float) to cents (int), normalize currency code
            cents = int(round(amount * 100))
            result = self._gateway.send_payment_request(cents, currency.upper())

            if result.get("status") == "SUCCESS":
                logger.info("Payment of %.2f %s processed", amount, currency)
                return True
            else:
                logger.warning("Payment failed: %s", result.get("message", "unknown"))
                return False
        except Exception as exc:
            logger.error("Adapter error processing %.2f %s: %s", amount, currency, exc)
            raise RuntimeError(f"Payment adapter failed: {exc}") from exc


# Example usage — client code knows only about PaymentProcessor
def checkout(processor: PaymentProcessor, total: float, currency: str) -> None:
    success = processor.process_payment(total, currency)
    if not success:
        raise ValueError("Checkout failed")

❌ BAD — Modifying the legacy class directly to match the new interface:

# ❌ BAD: Violates Open/Closed Principle — modifying external/legacy code
class BrokenLegacyGateway(LegacyPaymentGateway):
    def send_payment_request(self, value: int, code: str) -> dict[str, str]:
        # Someone had to modify this class just to fit our app...
        result = original_gateway.send_payment_request(value, code)
        if result.get("status") == "SUCCESS":
            return {"ok": True}  # Breaking the contract of LegacyPaymentGateway
        return {"ok": False}

✅ GOOD — Adapter wraps without modification:

# ✅ GOOD: Legacy code untouched; adapter translates between worlds
class CleanGatewayAdapter(PaymentProcessor):
    def __init__(self, gateway: LegacyPaymentGateway) -> None:
        self._gateway = gateway

    def process_payment(self, amount: float, currency: str) -> bool:
        cents = int(round(amount * 100))
        result = self._gateway.send_payment_request(cents, currency.upper())
        return result.get("status") == "SUCCESS"

Practical note: Use Adapter when integrating external libraries, migrating from one API version to another, or making a legacy module conform to a new interface contract. Choose Object Adapter (composition) over Class Adapter (inheritance) in Python — Python's multiple inheritance is fragile for this purpose.

Pattern 2: Strategy Pattern

Defines a family of interchangeable algorithms, encapsulates each one, and makes them swappable at runtime. The context holds a reference to a strategy interface and delegates to it instead of implementing the algorithm directly.

Use Strategy when you have multiple variants of an algorithm (e.g., sorting orders, pricing models, notification delivery methods) and you need to switch between them based on configuration, user choice, or runtime conditions — without bloating a single class with conditional logic.

from __future__ import annotations
import math
from typing import Protocol


class PricingStrategy(Protocol):
    """All pricing algorithms must implement this interface."""

    def calculate_price(self, base_price: float, quantity: int) -> float: ...


class FlatRatePricing:
    """Charges the standard list price per unit."""

    def calculate_price(self, base_price: float, quantity: int) -> float:
        if quantity <= 0:
            raise ValueError("Quantity must be positive")
        return round(base_price * quantity, 2)


class VolumeDiscountPricing:
    """Applies tiered volume discounts based on quantity thresholds."""

    TIERS: list[tuple[int, float]] = [
        (10, 0.05),   # 10+ units → 5% off
        (50, 0.10),   # 50+ units → 10% off
        (100, 0.15),  # 100+ units → 15% off
    ]

    def calculate_price(self, base_price: float, quantity: int) -> float:
        if quantity <= 0:
            raise ValueError("Quantity must be positive")

        discount = 0.0
        for threshold, rate in self.TIERS:
            if quantity >= threshold:
                discount = rate

        subtotal = base_price * quantity
        return round(subtotal * (1 - discount), 2)


class DynamicPricing:
    """Adjusts price based on demand multiplier and time-of-day factor."""

    def __init__(self, demand_multiplier: float = 1.0, time_factor: float = 1.0) -> None:
        self.demand_multiplier = max(0.5, min(demand_multiplier, 3.0))
        self.time_factor = max(0.7, min(time_factor, 2.0))

    def calculate_price(self, base_price: float, quantity: int) -> float:
        if quantity <= 0:
            raise ValueError("Quantity must be positive")
        adjusted = base_price * self.demand_multiplier * self.time_factor
        return round(adjusted * quantity, 2)


class PricingContext:
    """Delegates pricing decisions to a strategy. Can swap strategies at runtime."""

    def __init__(self, strategy: PricingStrategy) -> None:
        self._strategy = strategy

    @property
    def strategy(self) -> PricingStrategy:
        return self._strategy

    @strategy.setter
    def strategy(self, strategy: PricingStrategy) -> None:
        """Swap to a different pricing strategy at runtime."""
        self._strategy = strategy

    def calculate(self, base_price: float, quantity: int) -> float:
        return self._strategy.calculate_price(base_price, quantity)


# Example usage — context works with any strategy without knowing which one
def order_total(context: PricingContext, price: float, qty: int) -> float:
    return context.calculate(price, qty)

❌ BAD — Conditional logic instead of Strategy:

# ❌ BAD: New pricing type requires modifying the class every time
class BrokenOrderProcessor:
    def __init__(self, pricing_type: str = "flat") -> None:
        self._pricing_type = pricing_type

    def calculate(self, base_price: float, quantity: int) -> float:
        if self._pricing_type == "flat":
            return round(base_price * quantity, 2)
        elif self._pricing_type == "volume":
            discount = 0.10 if quantity >= 50 else 0.05
            return round(base_price * quantity * (1 - discount), 2)
        elif self._pricing_type == "dynamic":
            demand = 1.5  # magic constant — should be configurable
            return round(base_price * demand * quantity, 2)
        else:
            raise ValueError(f"Unknown pricing type: {self._pricing_type}")

✅ GOOD — Strategy pattern with interchangeable algorithms:

# ✅ GOOD: New strategy = new class, no modification to context
context = PricingContext(FlatRatePricing())
assert context.calculate(10.0, 5) == 50.0

context.strategy = VolumeDiscountPricing()
assert context.calculate(10.0, 60) == 540.0  # 10% volume discount applied

context.strategy = DynamicPricing(demand_multiplier=1.5)
assert context.calculate(10.0, 5) == 75.0

Practical note: Choose Strategy when behavior varies by configuration or external conditions and you want the selection logic decoupled from the execution logic. Use it instead of inheritance for runtime-swappable behavior; use inheritance (Template Method) for compile-time class hierarchies with shared step sequences.

Pattern 3: Observer Pattern

Establishes a one-to-many dependency between objects so that when one object (the subject) changes state, all its dependents (observers) are notified and updated automatically. Enables event-driven architecture with loose coupling between publishers and subscribers.

Use Observer when multiple components need to react to the same state change without knowing about each other directly. It is the backbone of event systems, pub/sub messaging, reactive UI frameworks, and domain events in DDD.

from __future__ import annotations
import weakref
from typing import Protocol
from abc import ABC


class Observer(ABC):
    """Base class for all observers."""

    def update(self, subject: str, event_data: dict) -> None: ...


class Subject:
    """Maintains a list of observers and notifies them of state changes.

    Uses weak references to prevent memory leaks from forgotten unsubscription.
    Thread-safe for basic add/remove/notify operations via internal lock-free
    copy-on-read pattern for the observer list.
    """

    def __init__(self) -> None:
        self._observers: list[weakref.ref[Observer]] = []
        self._state: dict[str, object] = {}

    @property
    def state(self) -> dict[str, object]:
        return dict(self._state)

    @state.setter
    def state(self, value: dict[str, object]) -> None:
        """Update state and notify all registered observers."""
        self._state = dict(value)  # Defensive copy
        self._notify_all()

    def register(self, observer: Observer) -> None:
        if not isinstance(observer, Observer):
            raise TypeError("Observer must implement the Observer protocol")
        ref = weakref.ref(observer)
        if ref not in self._observers:
            self._observers.append(ref)

    def unregister(self, observer: Observer) -> None:
        target_ref = weakref.ref(observer)
        self._observers = [r for r in self._observers if r != target_ref]

    def _notify_all(self) -> None:
        """Notify all live observers, filtering out garbage-collected ones."""
        live_observers: list[weakref.ref[Observer]] = []
        for ref in self._observers:
            obs = ref()
            if obs is not None:
                try:
                    obs.update(
                        subject=type(self).__name__,
                        event_data=dict(self._state),
                    )
                except Exception as exc:
                    # One observer's failure should not break others
                    import logging
                    logging.getLogger(__name__).warning(
                        "Observer %s failed: %s", obs, exc
                    )
                live_observers.append(ref)
            # Dead reference is silently removed on next notify cycle
        self._observers = live_observers


class PriceAlertObserver(Observer):
    """Watches for price changes and logs alerts."""

    def __init__(self, threshold: float) -> None:
        self.threshold = threshold

    def update(self, subject: str, event_data: dict) -> None:
        current = event_data.get("price", 0.0)
        if abs(current - self._prev_price) > self.threshold and hasattr(self, "_prev_price"):
            direction = "↑" if current > self._prev_price else "↓"
            print(f"[{subject}] Price {direction}: ${self._prev_price:.2f} → ${current:.2f}")
        self._prev_price = current


class EmailNotifierObserver(Observer):
    """Sends email notifications on specific events."""

    def __init__(self, email: str) -> None:
        self.email = email

    def update(self, subject: str, event_data: dict) -> None:
        if event_data.get("critical"):
            print(f"[{subject}] Email alert sent to {email}: "
                  f"Critical event — data={event_data}")


# Example usage — subject has no knowledge of concrete observer types
def demonstrate_observer() -> None:
    market = Subject()

    # Register observers
    alert = PriceAlertObserver(threshold=0.50)
    notifier = EmailNotifierObserver(email="trader@example.com")

    market.register(alert)
    market.register(notifier)

    # State changes trigger automatic notifications
    market.state = {"price": 150.0, "critical": False}
    market.state = {"price": 149.30, "critical": False}
    market.state = {"price": 125.00, "critical": True}

    # Unregister to stop receiving updates
    market.unregister(notifier)

❌ BAD — Tightly coupled state changes via direct calls:

# ❌ BAD: Every change requires manually calling every dependent
class BrokenPriceTracker:
    def __init__(self) -> None:
        self.alert_service = AlertService()
        self.email_service = EmailService()
        self.logger = Logger()
        self._price: float = 0.0

    @property
    def price(self) -> float:
        return self._price

    @price.setter
    def price(self, value: float) -> None:
        self._price = value
        # Every new consumer adds another direct call here — violates OCP
        self.alert_service.check_threshold(self._price)
        self.email_service.send_alert(self._price)
        self.logger.log_price_change(self._price)

✅ GOOD — Observer pattern with loose coupling:

# ✅ GOOD: Adding a new observer requires no changes to the subject
tracker = Subject()
tracker.register(PriceAlertObserver(threshold=1.0))
tracker.register(EmailNotifierObserver("ops@example.com"))
tracker.register(LoggerObserver())  # New observer — zero changes to Subject
tracker.state = {"price": 99.99}

Practical note: Use Observer when you have a broadcast notification requirement and multiple consumers should react independently. Guard against memory leaks by using weak references in the subject's observer registry. In async/await systems, consider async observers with asyncio.create_task to avoid blocking the subject's notify cycle.

Pattern 4: Command Pattern

Encapsulates a request as an object, thereby letting you parameterize clients with different requests, queue or log requests, and support undo/redo operations. Each command object captures all information needed to execute the action.

Use Command when you need to decouple the invoker (which triggers actions) from the receiver (which performs them), when you need to queue tasks for later execution, or when you need undo/redo capability — such as in editor applications, trading order systems, and workflow engines.

from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Any


class Command(ABC):
    """Abstract base for all commands."""

    @abstractmethod
    def execute(self) -> str: ...

    @abstractmethod
    def undo(self) -> str: ...


class OrderCommand(Command):
    """Encapsulates a trade order with full undo support.

    Stores the previous state so undo restores it exactly.
    Immutable command objects enable thread-safe command queues.
    """

    def __init__(self, account: Account, symbol: str, quantity: int, side: str) -> None:
        self.account = account
        self.symbol = symbol
        self.quantity = quantity
        self.side = side  # "BUY" or "SELL"

    def execute(self) -> str:
        if self.side == "BUY":
            result = self.account.buy(self.symbol, self.quantity)
        elif self.side == "SELL":
            result = self.account.sell(self.symbol, self.quantity)
        else:
            raise ValueError(f"Invalid side: {self.side}")
        return f"{self.side} {self.quantity} {self.symbol}: {result}"

    def undo(self) -> str:
        """Reverse the trade by doing the opposite operation."""
        opposite = "SELL" if self.side == "BUY" else "BUY"
        return self.account.sell(self.symbol, self.quantity) \
            if opposite == "SELL" else self.account.buy(self.symbol, self.quantity)


class Account:
    """Simple account with buy/sell operations and state tracking for undo."""

    def __init__(self, initial_cash: float = 10_000.0) -> None:
        self.cash = initial_cash
        self.positions: dict[str, int] = {}

    def buy(self, symbol: str, quantity: int) -> str:
        if quantity <= 0:
            raise ValueError("Quantity must be positive")
        price = 150.0 + hash(symbol) % 50  # Deterministic fake price
        cost = price * quantity
        if cost > self.cash:
            return f"Insufficient funds: need ${cost:.2f}, have ${self.cash:.2f}"
        self.cash -= cost
        self.positions[symbol] = self.positions.get(symbol, 0) + quantity
        return f"Purchased {quantity} {symbol} @ ${price:.2f}"

    def sell(self, symbol: str, quantity: int) -> str:
        if quantity <= 0:
            raise ValueError("Quantity must be positive")
        available = self.positions.get(symbol, 0)
        if available < quantity:
            return f"Insufficient {symbol} position: have {available}, want {quantity}"
        price = 155.0 + hash(symbol) % 45  # Deterministic fake price
        revenue = price * quantity
        self.cash += revenue
        self.positions[symbol] = available - quantity
        return f"Sold {quantity} {symbol} @ ${price:.2f}"


class CommandInvoker:
    """Manages command execution with undo/redo history.

    Maintains separate stacks for executed and undone commands,
    enabling full undo/redo support.
    """

    def __init__(self) -> None:
        self._history: list[Command] = []
        self._undone: list[Command] = []

    def execute_command(self, command: Command) -> str:
        result = command.execute()
        self._history.append(command)
        self._undone.clear()  # New action invalidates undone commands
        return result

    def undo(self) -> str:
        if not self._history:
            return "Nothing to undo"
        command = self._history.pop()
        result = command.undo()
        self._undone.append(command)
        return f"Undo: {result}"

    def redo(self) -> str:
        if not self._undone:
            return "Nothing to redo"
        command = self._undone.pop()
        # Re-execute the original command
        result = command.execute()
        self._history.append(command)
        return f"Redo: {result}"


# Example usage — invoker is completely decoupled from command details
def demonstrate_command() -> None:
    account = Account(initial_cash=10_000.0)
    invoker = CommandInvoker()

    buy_cmd = OrderCommand(account, "AAPL", 10, "BUY")
    print(invoker.execute_command(buy_cmd))   # BUY 10 AAPL: ...
    print(invoker.undo())                      # Undo: ... (sells back)
    print(invoker.redo())                      # Redo: ... (buys again)

❌ BAD — Direct method calls with no encapsulation or undo:

# ❌ BAD: No abstraction layer, no undo capability
def execute_trade(account: Account, symbol: str, qty: int, side: str) -> None:
    if side == "BUY":
        account.buy(symbol, qty)
    else:
        account.sell(symbol, qty)
    # What if the user clicks the button twice? No idempotency or undo.


# ❌ BAD: Coupled invoker that knows about concrete operations
class BrokenUIInvoker:
    def __init__(self, account: Account) -> None:
        self.account = account

    def buy_button_clicked(self, symbol: str, qty: int) -> None:
        self.account.buy(symbol, qty)  # Tightly coupled to this exact operation

✅ GOOD — Decoupled command with undo/redo:

# ✅ GOOD: Invoker manages history; commands are reusable and testable in isolation
invoker = CommandInvoker()
buy_aapl = OrderCommand(Account(), "AAPL", 5, "BUY")
sell_aapl = OrderCommand(buy_aapl.account, "AAPL", 2, "SELL")

print(invoker.execute_command(buy_aapl))  # Execute first command
print(invoker.execute_command(sell_aapl))  # Queue second command
print(invoker.undo())                       # Undo last: sell reversed → buy restored
print(invoker.undo())                       # Undo again: buy reversed → cash restored

Practical note: Use Command when you need to parameterize operations, support undo/redo queues, or decouple the UI layer from business logic. For persistent command queues (e.g., message brokers), serialize the command parameters as data and rehydrate them on the worker side rather than serializing Python objects directly.

Pattern 5: Facade Pattern

Provides a simplified, unified interface to a complex subsystem. The facade defines a higher-level interface that makes the subsystem easier to use by hiding complexity and dependencies.

Use Facade when an application's API is overly complex, when you want to decouple a high-level client from a maze of library classes, or when you need a single entry point for initializing a subsystem (e.g., bootstrapping a trading platform, starting an ETL pipeline).

from __future__ import annotations
import logging
from typing import Optional

logger = logging.getLogger(__name__)


# Complex subsystem components — the client should NOT interact with these directly
class DataFeed:
    """Low-level market data feed management."""

    def connect(self, endpoint: str) -> None:
        logger.info("Connecting to data feed at %s", endpoint)

    def subscribe(self, symbol: str) -> dict:
        return {"symbol": symbol, "price": 150.0 + hash(symbol) % 100}

    def disconnect(self) -> None:
        logger.info("Disconnected from data feed")


class OrderManager:
    """Low-level order management."""

    def create_order(self, symbol: str, side: str, qty: int, price: Optional[float] = None) -> dict:
        return {"order_id": hash(f"{symbol}{side}{qty}") % 1_000_000,
                "status": "submitted", "filled": False}

    def cancel_order(self, order_id: int) -> bool:
        return True


class RiskEngine:
    """Low-level risk checking."""

    def check_order(self, symbol: str, side: str, qty: int, cash: float) -> tuple[bool, str]:
        estimated_cost = 150.0 * qty
        if estimated_cost > cash:
            return False, f"Insufficient funds for {side} order"
        return True, "Risk check passed"


class PortfolioTracker:
    """Low-level portfolio tracking."""

    def update(self, symbol: str, side: str, qty: int, price: float) -> None:
        logger.info("Portfolio updated: %s %s %d @ %.2f", symbol, side, qty, price)


class TradingSystemFacade:
    """Simplified unified interface to the entire trading subsystem.

    Clients interact only with this facade — they never need to know about
    DataFeed, OrderManager, RiskEngine, or PortfolioTracker individually.
    """

    def __init__(self) -> None:
        self._data_feed = DataFeed()
        self._order_manager = OrderManager()
        self._risk_engine = RiskEngine()
        self._portfolio_tracker = PortfolioTracker()
        self._connected = False

    def start(self, feed_endpoint: str) -> str:
        """Initialize the trading system with a single call."""
        self._data_feed.connect(feed_endpoint)
        self._connected = True
        return "Trading system initialized"

    def place_order(self, symbol: str, side: str, qty: int, cash: float = 10_000.0) -> dict:
        """Place a trade through the unified interface.

        Internally subscribes to data, checks risk, creates order, and tracks portfolio —
        all without the caller needing to manage any of those steps.
        """
        if not self._connected:
            raise RuntimeError("Trading system not started — call start() first")

        # Step 1: Check risk
        allowed, message = self._risk_engine.check_order(symbol, side, qty, cash)
        if not allowed:
            return {"error": message}

        # Step 2: Subscribe to price data
        data = self._data_feed.subscribe(symbol)

        # Step 3: Create order
        order = self._order_manager.create_order(symbol, side, qty, data.get("price"))

        # Step 4: Update portfolio tracking
        if order.get("status") == "submitted":
            self._portfolio_tracker.update(symbol, side, qty, data["price"])

        return order

    def stop(self) -> str:
        """Clean up resources with one call."""
        if self._connected:
            self._data_feed.disconnect()
            self._connected = False
        return "Trading system stopped"


# Example usage — client has zero knowledge of subsystem internals
def demonstrate_facade() -> None:
    trading = TradingSystemFacade()

    # One call to initialize the entire complex subsystem
    print(trading.start("wss://market-data.example.com/stream"))

    # One call to place an order, hiding 4 internal steps
    result = trading.place_order("AAPL", "BUY", 10)
    print(result)

    # One call to clean up
    print(trading.stop())

❌ BAD — Client directly orchestrating the complex subsystem:

# ❌ BAD: Client must know about every subsystem component
class BrokenClient:
    def place_trade(self, symbol: str, side: str, qty: int) -> None:
        data = DataFeed()
        data.connect("wss://feed.example.com")

        risk = RiskEngine()
        ok, msg = risk.check_order(symbol, side, qty, 10_000.0)
        if not ok:
            return

        orders = OrderManager()
        order = orders.create_order(symbol, side, qty)

        portfolio = PortfolioTracker()
        portfolio.update(symbol, side, qty, 150.0)

✅ GOOD — Client interacts with a single simplified facade:

# ✅ GOOD: Client code is clean and focused on intent, not infrastructure
system = TradingSystemFacade()
system.start("wss://market-data.example.com/stream")
system.place_order("AAPL", "BUY", 10)
system.stop()

Practical note: Use Facade when you want to provide a simplified API without restricting the underlying subsystem's capabilities. Clients can always access the underlying components directly if needed. A common anti-pattern is turning a Facade into a God Object — keep it as a thin delegator, not a container of business logic.

Pattern 6: Template Method Pattern

Defines the skeleton of an algorithm in a base class, deferring specific steps to subclasses. Subclasses override particular steps without changing the overall algorithm structure. The base class controls the order and guarantees that certain steps are always executed.

Use Template Method when you have multiple algorithms that share the same high-level structure but differ in specific steps, such as report generation (different data sources, same formatting pipeline), data processing pipelines (same validation → transform → load sequence, different implementations), or testing frameworks.

from __future__ import annotations
from abc import ABC, abstractmethod
import logging

logger = logging.getLogger(__name__)


class ReportGenerator(ABC):
    """Base class defining the skeleton of report generation.

    The template method generate_report() is final — subclasses cannot reorder steps.
    Subclasses override only the steps that differ between report types.
    """

    def generate_report(self) -> str:
        """The template method: controls algorithm structure."""
        self._validate_inputs()
        self._fetch_data()
        self._transform_data()
        results = self._format_output()
        self._save_report(results)
        return results

    # Template steps that subclasses MAY override — with sensible defaults
    def _validate_inputs(self) -> None:
        """Default validation; override to add domain-specific checks."""
        logger.debug("Using default input validation")

    @abstractmethod
    def _fetch_data(self) -> list[dict]:
        """Must be implemented by subclasses — data source varies."""
        ...

    def _transform_data(self, data: list[dict] | None = None) -> list[dict]:
        """Default transformation; override for custom processing."""
        if data is None:
            data = self._fetch_data()
        return [row for row in data if row.get("value") is not None]

    @abstractmethod
    def _format_output(self, transformed_data: list[dict]) -> str: ...

    def _save_report(self, content: str) -> None:
        """Default persistence; override to change storage target."""
        logger.info("Report saved (%d characters)", len(content))


class SalesReport(ReportGenerator):
    """Generates sales reports from a CSV data source."""

    def _validate_inputs(self) -> None:
        if not hasattr(self, "region") or not self.region:
            raise ValueError("Region is required for sales reports")

    def _fetch_data(self) -> list[dict]:
        # In production, this reads from a database or CSV file
        return [
            {"product": "Widget", "value": 1200},
            {"product": "Gadget", "value": 3400},
            {"product": "Doohickey", "value": None},  # Should be filtered out
        ]

    def _format_output(self, transformed_data: list[dict]) -> str:
        lines = ["=== SALES REPORT ===", f"Region: {self.region}"]
        for row in transformed_data:
            lines.append(f"  {row['product']}: ${row['value']}")
        return "\n".join(lines)

    def _save_report(self, content: str) -> None:
        logger.info("SALES REPORT saved to /reports/sales/%s", self.region)


class PerformanceReport(ReportGenerator):
    """Generates performance reports from API metrics."""

    def __init__(self, endpoint: str) -> None:
        self.endpoint = endpoint

    def _validate_inputs(self) -> None:
        if not self.endpoint.startswith(("http://", "https://")):
            raise ValueError(f"Invalid endpoint URL: {self.endpoint}")

    def _fetch_data(self) -> list[dict]:
        # In production, this calls the API
        return [
            {"metric": "latency_p99", "value": 250},
            {"metric": "error_rate", "value": 0.02},
            {"metric": "throughput", "value": None},  # Filtered out
        ]

    def _format_output(self, transformed_data: list[dict]) -> str:
        lines = ["=== PERFORMANCE REPORT ==="]
        for row in transformed_data:
            lines.append(f"  {row['metric']}: {row['value']}")
        return "\n".join(lines)


# Example usage — the template method structure is fixed, only data differs
def demonstrate_template_method() -> None:
    sales = SalesReport()
    sales.region = "North America"
    print(sales.generate_report())

    perf = PerformanceReport(endpoint="https://api.example.com/metrics")
    print(perf.generate_report())

❌ BAD — Duplicate algorithm code in every subclass:

# ❌ BAD: Every subclass reimplements the entire pipeline — violation of DRY
class BrokenSalesReport:
    def generate(self) -> str:
        self._validate()           # Redundant implementation
        data = self._fetch()       # Different source, same structure
        cleaned = [d for d in data if d.get("value") is not None]  # Copy-paste
        formatted = f"Sales:\n" + "\n".join(f"  {v}" for v in cleaned)
        self._save(formatted)      # Another copy of the save logic
        return formatted

    # Same validate, _save methods duplicated across every report type...


class BrokenPerfReport:
    def generate(self) -> str:
        self._validate()           # Nearly identical validation
        data = self._fetch()       # Different source again
        cleaned = [d for d in data if d.get("value") is not None]  # Exact copy
        formatted = f"Perf:\n" + "\n".join(f"  {v}" for v in cleaned)
        self._save(formatted)      # Same save logic again
        return formatted

✅ GOOD — Template Method enforces structure once, varies only what differs:

# ✅ GOOD: generate_report() is defined exactly once in the base class
# Each subclass only overrides the steps that actually differ
report = PerformanceReport(endpoint="https://metrics.internal/v1")
print(report.generate_report())  # validate → fetch → transform → format → save

Practical note: Use Template Method when the algorithm structure is invariant but some steps vary. Use Strategy (behavioral pattern) instead when the entire algorithm varies and should be swappable at runtime — Template Method's variation points are locked at compile time via inheritance, while Strategy swaps entire algorithms at runtime.

Pattern 7: Decorator Pattern

Attaches additional responsibilities to an object dynamically. Decorators provide a flexible alternative to subclassing for extending functionality. The decorator wraps the original object and adds behavior before/after delegating to it.

Use Decorator when you need to add behaviors (logging, caching, authent

…(truncated)