SimulatorCost can't use arbitrary estimators as nested dependents; subclassing is difficult

[harveechen]

class SimlatorCost's member function simulate call self.values_in_time to update parameter value, but when the cost has member field that need to update, it will make a mistake.

[enzbus]

Hello, thank you. You're referring to line 443 of costs.py; I know the logic is tricky there but can you please explain which specific usecase broke for you? I'm quite confident it works fine with the derived classes implemented in Cvxportfolio, I would need to know more about your usecase to see what needs fixing.

[enzbus]

PS you implemented a derived cost and you have custom logic inside your simulate method you want executed? I guess best solution then would be to call super().simulate before or after your logic. Does that make sense?

[enzbus]

Ok sorry now I see your point; you're probably right let me try the change and will cut a revision if necessary. Thanks!

[harveechen]

Thx for reply!
I will give a example to show my case.

class MyCustomCost (SimulatorCost):
    def __init__(self, a=5/10_000, b = 1 / 10_000):
        self.a = None if a is None else DataEstimator(a)
        self.b = None if b is None else DataEstimator(b)
        self._first_term_multiplier = None
        self._second_term_multiplier = None
        super().__init__()

    def initialize_estimator(self, universe, **kwargs):
        if self.a is not None:
            self._first_term_multiplier = cp.Parameter(
                len(universe)-1, nonneg=True)
        if self.b is not None:
            self._second_term_multiplier = cp.Parameter(
                len(universe)-1, nonneg=True)
        super().initialize_estimator(universe=universe, **kwargs) # initial w w_plus ...

    def values_in_time(self, **kwargs):
        if self.a is not None:
            self._first_term_multiplier.value = np.ones(self._first_term_multiplier.size) * self.a.current_value
        
        if self.b is not None:
            self._second_term_multiplier.value = np.ones(self._second_term_multiplier.size) * self.b.current_value

    def compile_to_cvxpy(self, z, **kwargs):
        expression = 0
        if self.a is not None:
            expression += cp.abs[z[:-1]] @ self._first_term_multiplier
            assert expression.is_convex()
        if self.b is not None:
            expression += cp.abs[z[:-1]] @ self._second_term_multiplier
            assert expression.is_convex()
        return expression

# send cost to simulator creator, MyCustomCost is like TransactionCost but with default fee
# and will use self.xx.current_value to populate 
simulator = cvx.MarketSimulator(
        universe=UNIVERSE,
        costs = [MyCustomCost])

# then when the simulate is called, it will call values_in_time, however, self.a.current_value will be None if self.a is not update by 
# calling self.a.values_in_time. That what I say it need to call values_in_time_recursive in simulate super function.

[enzbus]

I guess I'll refactor the SimulatorCost object when I have some time and document the interface better; in any case using values_in_time_recursive doesn't work. I tried your code, and I can't reproduce the bug you report. Here's a minimal modification of the code you gave, in order to run it:

import cvxportfolio as cvx
import cvxpy as cp
import numpy as np

class MyCustomCost (cvx.costs.SimulatorCost):
    def __init__(self, a=5/10_000, b = 1 / 10_000):
        self.a = None if a is None else cvx.estimator.DataEstimator(a)
        self.b = None if b is None else cvx.estimator.DataEstimator(b)
        self._first_term_multiplier = None
        self._second_term_multiplier = None
        super().__init__()

    def initialize_estimator(self, universe, **kwargs):
        if self.a is not None:
            self._first_term_multiplier = cp.Parameter(
                len(universe)-1, nonneg=True)
        if self.b is not None:
            self._second_term_multiplier = cp.Parameter(
                len(universe)-1, nonneg=True)
        super().initialize_estimator(universe=universe, **kwargs) # initial w w_plus ...

    def values_in_time(self, **kwargs):
        if self.a is not None:
            self._first_term_multiplier.value = np.ones(self._first_term_multiplier.size) * self.a.current_value
        
        if self.b is not None:
            self._second_term_multiplier.value = np.ones(self._second_term_multiplier.size) * self.b.current_value

    def compile_to_cvxpy(self, z, **kwargs):
        expression = 0
        if self.a is not None:
            expression += cp.abs(z[:-1]) @ self._first_term_multiplier
            assert expression.is_convex()
        if self.b is not None:
            expression += cp.abs(z[:-1]) @ self._second_term_multiplier
            assert expression.is_convex()
        return expression

# send cost to simulator creator, MyCustomCost is like TransactionCost but with default fee
# and will use self.xx.current_value to populate 

UNIVERSE = ['AAPL', 'GOOG']
simulator = cvx.MarketSimulator(
        universe=UNIVERSE,
        costs = [MyCustomCost])

print(simulator.backtest(cvx.Uniform(), start_time='2025-01-01'))

And it works fine, here's the output

[harveechen]

Sorry, I recheck the code and find that cost of simulator will call simulate_recursive which will call self.a simulate function. I will close this issue

[enzbus]

Let me reopen this. I remembered now how I designed that logic and it does break if you push it enough. Your example works because DataEstimator also inherits from SimulatorEstimator, it has a simulate method that itself calls its own values_in_time. If you were to use some other estimator, for example one of the CvxpyExpressionEstimator used by the rest of the library, it wouldn't work. As you correctly noticed, values_in_time_recursive is never called so the evaluation tree is not executed. I will put it as a work item, plus documenting the behavior for ease of subclassing.