beacon.fund

Investable vehicles: IndexFund composes an IndexCalculator and a BacktestEngine to track an index, with management-fee accrual; ETF extends it with a ticker, creation-unit size, market-price simulation, and tracking-performance analysis.

fund

The init.py for the 'fund' module.

This module models financial funds, particularly ETFs and index funds.

IndexFund

IndexFund(
    fund_id: str,
    target_index_definition: IndexDefinition,
    index_agent: IndexCalculator,
    portfolio: Portfolio,
    data_provider: DataFetcher,
    management_fee_bps: int = 0,
)

An index fund that tracks a target index.

The fund delegates the whole pipeline — target weight calculation and the simulated tracking portfolio — to :class:~beacon.backtest.main.Backtest, the front door composing :class:~beacon.index.calculation.IndexCalculator and :class:~beacon.backtest.engine.BacktestEngine (BN-161). It contains no buy/sell logic of its own — rebalancing and portfolio accounting are delegated entirely to the backtest engine.

Initializes an IndexFund.

Parameters:

Name	Type	Description	Default
fund_id	str	A unique identifier for the fund.	required
target_index_definition	IndexDefinition	The definition of the index the fund aims to track.	required
index_agent	IndexCalculator	The IndexCalculator used to compute the target index's weight schedule.	required
portfolio	Portfolio	The Portfolio object seeding the fund's capital. Its cash balance is used as the backtest engine's initial capital; the fund no longer mutates this portfolio directly.	required
data_provider	DataFetcher	DataFetcher instance for market data.	required
management_fee_bps	int	The annual management fee in basis points (e.g., 10 bps = 0.1%).	0

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/fund/base.py

def __init__(self,
             fund_id: str,
             target_index_definition: IndexDefinition,
             index_agent: IndexCalculator,
             portfolio: Portfolio,
             data_provider: DataFetcher,
             management_fee_bps: int = 0):
    """
    Initializes an IndexFund.

    Args:
        fund_id: A unique identifier for the fund.
        target_index_definition: The definition of the index the fund aims to track.
        index_agent: The IndexCalculator used to compute the target index's
                     weight schedule.
        portfolio: The Portfolio object seeding the fund's capital. Its cash
                   balance is used as the backtest engine's initial capital;
                   the fund no longer mutates this portfolio directly.
        data_provider: DataFetcher instance for market data.
        management_fee_bps: The annual management fee in basis points (e.g., 10 bps = 0.1%).
    """
    if not fund_id:
        raise ValueError("fund_id cannot be empty.")
    if not target_index_definition:
        raise ValueError("target_index_definition must be provided.")
    if not index_agent:
        raise ValueError("index_agent must be provided.")
    if not portfolio:
        raise ValueError("portfolio must be provided.")
    if not data_provider:
        raise ValueError("data_provider must be provided.")
    if management_fee_bps < 0:
        raise ValueError("management_fee_bps cannot be negative.")

    self.fund_id: str = fund_id
    self.target_index_definition: IndexDefinition = target_index_definition
    self.index_agent: IndexCalculator = index_agent
    self.portfolio: Portfolio = portfolio
    self.data_provider: DataFetcher = data_provider
    self.management_fee_bps: int = management_fee_bps  # e.g., 20 for 0.20%

    # Cached outputs of the composed calculator + engine pipeline.
    self._index_result: IndexResult | None = None
    self._backtest_result: BacktestResult | None = None

index_result property

index_result: IndexResult | None

The target :class:IndexResult from the most recent run, if any.

backtest_result property

backtest_result: BacktestResult | None

The :class:BacktestResult from the most recent run, if any.

run_backtest

run_backtest(
    start_date: str | None = None,
    end_date: str | None = None,
    transaction_cost_bps: float = 0.0,
) -> BacktestResult

Compute target weights and simulate the tracking portfolio.

Delegates the whole calculate-then-simulate composition to :class:~beacon.backtest.main.Backtest (BN-161), which fingerprints the calculation, reuses a cached IndexResult when the data source allows it, and hands the schedule to a backtest engine that manages its own portfolio. The resulting :class:BacktestResult is cached on the fund and returned.

Parameters:

Name	Type	Description	Default
start_date	str | None	First simulation date (YYYY-MM-DD). Defaults to the target index's base date.	None
end_date	str | None	Last simulation date (YYYY-MM-DD). Required.	None
transaction_cost_bps	float	Trading cost applied by the engine to each trade's notional. Distinct from the fund's management fee.	0.0

Returns:

Type	Description
BacktestResult	The BacktestResult produced by the engine.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/fund/base.py

def run_backtest(self,
                 start_date: str | None = None,
                 end_date: str | None = None,
                 transaction_cost_bps: float = 0.0) -> BacktestResult:
    """Compute target weights and simulate the tracking portfolio.

    Delegates the whole calculate-then-simulate composition to
    :class:`~beacon.backtest.main.Backtest` (BN-161), which fingerprints
    the calculation, reuses a cached IndexResult when the data source
    allows it, and hands the schedule to a backtest engine that manages
    its own portfolio. The resulting :class:`BacktestResult` is cached on
    the fund and returned.

    Args:
        start_date: First simulation date (YYYY-MM-DD). Defaults to the
            target index's base date.
        end_date: Last simulation date (YYYY-MM-DD). Required.
        transaction_cost_bps: Trading cost applied by the engine to each
            trade's notional. Distinct from the fund's management fee.

    Returns:
        The BacktestResult produced by the engine.
    """
    if end_date is None:
        raise ValueError("end_date must be provided to run the fund backtest.")

    base_date = self.target_index_definition.base_date
    start = start_date or base_date.strftime('%Y-%m-%d')

    logger.info(
        f"Fund '{self.fund_id}': computing target weights for "
        f"'{self.target_index_definition.index_name}' from {start} to {end_date}."
    )

    backtest = Backtest(initial_capital=self.portfolio.cash_balance,
                        transaction_cost_bps=transaction_cost_bps,
                        price_column=self.index_agent.price_column,
                        data_provider=self.data_provider)
    self._backtest_result = backtest.run(self.target_index_definition,
                                         start=start,
                                         end=end_date)

    # The calculation the run tracked, kept for the fund's own accessor.
    book = self._backtest_result.index.target
    self._index_result = book.source if book is not None else None

    logger.info(
        f"Fund '{self.fund_id}': backtest complete "
        f"({len(self._backtest_result.trading_nav)} days, "
        f"{len(self._backtest_result.portfolio.transactions)} transactions)."
    )
    return self._backtest_result

rebalance_to_index

rebalance_to_index(current_date: Timestamp) -> None

Align the fund's tracking portfolio with the target index.

Thin wrapper that ensures the composed calculator + engine pipeline has been run through current_date. All weight computation is delegated to the :class:IndexCalculator and all trading to the :class:BacktestEngine; this class performs no buy/sell logic itself.

Parameters:

Name	Type	Description	Default
current_date	Timestamp	The date through which to simulate.	required

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/fund/base.py

def rebalance_to_index(self,
                       current_date: pd.Timestamp) -> None:
    """Align the fund's tracking portfolio with the target index.

    Thin wrapper that ensures the composed calculator + engine pipeline has
    been run through *current_date*. All weight computation is delegated to
    the :class:`IndexCalculator` and all trading to the
    :class:`BacktestEngine`; this class performs no buy/sell logic itself.

    Args:
        current_date: The date through which to simulate.
    """
    self._ensure_backtest(pd.Timestamp(current_date))

calculate_nav

calculate_nav(current_date: Timestamp) -> float

Return the fund's Net Asset Value as of current_date.

The gross NAV is read from the backtest-engine-managed portfolio; the accrued management fee is then deducted.

Parameters:

Name	Type	Description	Default
current_date	Timestamp	The date for which to calculate NAV.	required

Returns:

Type	Description
float	The fee-adjusted Net Asset Value.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/fund/base.py

def calculate_nav(self,
                  current_date: pd.Timestamp) -> float:
    """Return the fund's Net Asset Value as of *current_date*.

    The gross NAV is read from the backtest-engine-managed portfolio; the
    accrued management fee is then deducted.

    Args:
        current_date: The date for which to calculate NAV.

    Returns:
        The fee-adjusted Net Asset Value.
    """
    ts = pd.Timestamp(current_date)
    self._ensure_backtest(ts)

    if self._backtest_result is None or self._backtest_result.trading_nav.empty:
        # Date precedes the simulation window — only seed capital exists.
        return float(self.portfolio.cash_balance)

    # The trading NAV, deliberately: the fee accrues over elapsed
    # NAV-series days, and the day-zero row (decision 11) is a starting
    # fact, not an elapsed day. Reading it here would shift every day
    # count by one and change accrued fees; excluding it keeps them
    # bit-identical to the pre-redesign series -- the fund tests are the
    # proof.
    nav_series = self._backtest_result.trading_nav
    on_or_before = nav_series.index[nav_series.index <= ts]
    if len(on_or_before) == 0:
        return float(self.portfolio.cash_balance)

    as_of = on_or_before[-1]
    gross_nav = float(nav_series.loc[as_of])
    elapsed_days = nav_series.index.get_loc(as_of)  # 0 on the first day

    net_nav = self._apply_management_fee(gross_nav, elapsed_days)
    logger.debug(
        f"NAV for fund '{self.fund_id}' on {ts.strftime('%Y-%m-%d')}: "
        f"gross={gross_nav:.2f}, net={net_nav:.2f}"
    )
    return net_nav

ETF

ETF(
    fund_id: str,
    etf_ticker: str,
    target_index_definition: IndexDefinition,
    index_agent: IndexCalculator,
    portfolio: Portfolio,
    data_provider: DataFetcher,
    management_fee_bps: int = 0,
    creation_unit_size: int = 50000,
)

Bases: IndexFund

Represents an Exchange Traded Fund (ETF), which is a type of IndexFund with additional characteristics like market price and creation/redemption units.

Initializes an ETF.

Parameters:

Name	Type	Description	Default
fund_id	str	A unique identifier for the fund.	required
etf_ticker	str	The market ticker symbol for the ETF.	required
target_index_definition	IndexDefinition	The definition of the index the ETF tracks.	required
index_agent	IndexCalculator	Calculation agent for the target index.	required
portfolio	Portfolio	The Portfolio object representing the ETF's holdings.	required
data_provider	DataFetcher	DataFetcher for market data.	required
management_fee_bps	int	Annual management fee in basis points.	0
creation_unit_size	int	The number of ETF shares in a creation/redemption unit.	50000

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/fund/etf.py

def __init__(self,
             fund_id: str,
             etf_ticker: str,
             target_index_definition: IndexDefinition,
             index_agent: IndexCalculator,
             portfolio: Portfolio,
             data_provider: DataFetcher,
             management_fee_bps: int = 0,
             creation_unit_size: int = 50000): # Typical size of a creation unit
    """
    Initializes an ETF.

    Args:
        fund_id: A unique identifier for the fund.
        etf_ticker: The market ticker symbol for the ETF.
        target_index_definition: The definition of the index the ETF tracks.
        index_agent: Calculation agent for the target index.
        portfolio: The Portfolio object representing the ETF's holdings.
        data_provider: DataFetcher for market data.
        management_fee_bps: Annual management fee in basis points.
        creation_unit_size: The number of ETF shares in a creation/redemption unit.
    """
    super().__init__(fund_id=fund_id,
                     target_index_definition=target_index_definition,
                     index_agent=index_agent,
                     portfolio=portfolio,
                     data_provider=data_provider,
                     management_fee_bps=management_fee_bps)
    if not etf_ticker:
        raise ValueError("etf_ticker cannot be empty.")
    if creation_unit_size <= 0:
        raise ValueError("creation_unit_size must be positive.")

    self.etf_ticker: str = etf_ticker
    self.creation_unit_size: int = creation_unit_size
    self.market_price: float | None = None # Simulated or actual market price

simulate_market_price

simulate_market_price(
    current_date: Timestamp,
    market_factors: dict[str, Any] | None = None,
) -> float

Simulates the ETF's market price based on its NAV and other market factors. (Future Scope: Initial focus on NAV tracking implies market price might closely follow NAV, or be supplied externally if backtesting against actual ETF data).

For a basic simulation, market price might be NAV plus some noise or bid-ask spread. This is a placeholder for more sophisticated modeling.

Parameters:

Name	Type	Description	Default
current_date	Timestamp	The date for which to simulate the price.	required
market_factors	dict[str, Any] | None	A dictionary of factors that might influence the price (e.g., market sentiment, liquidity, bid-ask spread).	None

Returns:

Type	Description
float	The simulated market price of the ETF.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/fund/etf.py

def simulate_market_price(self,
                          current_date: pd.Timestamp,
                          market_factors: dict[str, Any] | None = None) -> float:
    """
    Simulates the ETF's market price based on its NAV and other market factors.
    (Future Scope: Initial focus on NAV tracking implies market price might closely follow NAV,
     or be supplied externally if backtesting against actual ETF data).

    For a basic simulation, market price might be NAV plus some noise or bid-ask spread.
    This is a placeholder for more sophisticated modeling.

    Args:
        current_date: The date for which to simulate the price.
        market_factors: A dictionary of factors that might influence the price
                        (e.g., market sentiment, liquidity, bid-ask spread).

    Returns:
        The simulated market price of the ETF.
    """
    nav_per_share = self.calculate_nav(current_date) # Assuming NAV is total value.
    # If NAV per share requires number of ETF shares outstanding:
    # num_etf_shares = self.portfolio.get_total_shares() # Needs implementation if ETF
    # shares tracked
    # nav_per_share = self.calculate_nav(current_date) / num_etf_shares if num_etf_shares
    # else nav_per_share

    # Simplistic simulation: market price = NAV (perfect tracking for now)
    self.market_price = nav_per_share
    logger.debug(f"Simulated market price for ETF '{self.etf_ticker}' on "
                 f"{current_date.strftime('%Y-%m-%d')}: {self.market_price:.2f} (based on NAV)")
    # Add more complex logic here later, e.g., premium/discount simulation
    return self.market_price

get_tracking_performance

get_tracking_performance(
    result: BacktestResult,
) -> dict[str, float | str]

Calculate tracking metrics from a completed backtest.

Compares the backtest's trading_nav against the tracked index's index_levels using the tracking methods built into :class:~beacon.backtest.result.BacktestResult. The result must carry an index_result for the comparison to be possible.

Parameters:

Name	Type	Description	Default
result	BacktestResult	A BacktestResult produced by tracking this ETF's index. It already contains both the portfolio NAV and the target index.	required

Returns:

Type	Description
dict[str, float | str]	A dictionary with float tracking_error and
dict[str, float | str]	tracking_difference entries. If the result has no target index
dict[str, float | str]	to compare against, a single error entry is returned instead,
dict[str, float | str]	whose value is an explanatory string — hence the float | str
dict[str, float | str]	value type.

Raises:

Type	Description
ValueError	If result is None.

Source code in build/cache/py-beacon-2c9c3936c65abdb6b8403c50a023f58355be30ee/src/beacon/fund/etf.py

def get_tracking_performance(self,
                             result: BacktestResult) -> dict[str, float | str]:
    """Calculate tracking metrics from a completed backtest.

    Compares the backtest's ``trading_nav`` against the tracked index's
    ``index_levels`` using the tracking methods built into
    :class:`~beacon.backtest.result.BacktestResult`. The *result* must carry
    an ``index_result`` for the comparison to be possible.

    Args:
        result: A BacktestResult produced by tracking this ETF's index. It
            already contains both the portfolio NAV and the target index.

    Returns:
        A dictionary with float ``tracking_error`` and
        ``tracking_difference`` entries. If the result has no target index
        to compare against, a single ``error`` entry is returned instead,
        whose value is an explanatory string — hence the ``float | str``
        value type.

    Raises:
        ValueError: If *result* is None.
    """
    if result is None:
        raise ValueError("A BacktestResult must be provided.")

    logger.info(f"Calculating tracking performance for ETF '{self.etf_ticker}'.")

    tracking_err = result.get_tracking_error()
    tracking_diff = result.get_tracking_difference()

    if tracking_err is None or tracking_diff is None:
        logger.error(
            f"BacktestResult for ETF '{self.etf_ticker}' has no target index "
            "to compare against."
        )
        return {"error": "BacktestResult has no target index for tracking comparison."}

    logger.info(
        f"Tracking performance for '{self.etf_ticker}': "
        f"TE={tracking_err:.4f}, TD={tracking_diff:.4f}"
    )
    return {
        "tracking_error": tracking_err,
        "tracking_difference": tracking_diff,
    }