Yuxuan Yan
51 lines
1.4 KiB
Python
51 lines
1.4 KiB
Python
"""Trading cost models for portfolio execution simulation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass
|
|
from typing import Mapping
|
|
|
|
import numpy as np
|
|
|
|
|
|
class CostModel(ABC):
|
|
"""Interface for per-name execution cost models."""
|
|
|
|
@abstractmethod
|
|
def compute(
|
|
self,
|
|
traded_shares: np.ndarray,
|
|
execution_price: np.ndarray,
|
|
side: np.ndarray,
|
|
date,
|
|
metadata: Mapping[str, object] | None = None,
|
|
) -> np.ndarray:
|
|
"""Return per-name trading cost in yuan."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SimpleProportionalCostModel(CostModel):
|
|
"""Simplified open-execution proportional cost model.
|
|
|
|
Slippage is represented as an additional cash cost. The execution price is
|
|
not adjusted by slippage, which avoids double-counting.
|
|
"""
|
|
|
|
cost_bps: float = 0.0
|
|
slippage_bps: float = 0.0
|
|
|
|
def compute(
|
|
self,
|
|
traded_shares: np.ndarray,
|
|
execution_price: np.ndarray,
|
|
side: np.ndarray,
|
|
date,
|
|
metadata: Mapping[str, object] | None = None,
|
|
) -> np.ndarray:
|
|
shares = np.asarray(traded_shares, dtype=np.float64)
|
|
price = np.asarray(execution_price, dtype=np.float64)
|
|
open_price = np.where(np.isfinite(price), price, 0.0)
|
|
trade_value = np.abs(shares * open_price)
|
|
return trade_value * (self.cost_bps + self.slippage_bps) / 1e4
|