Spaces:
Sleeping
Sleeping
File size: 1,722 Bytes
4f55ca2 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 |
from abc import ABC, abstractmethod
from typing import Dict
import numpy as np
from ibydmt.utils import _get_cls, _register_cls
class Wealth(ABC):
def __init__(self, config):
self.significance_level = config.significance_level
self.rejected = False
@abstractmethod
def update(self, payoff):
pass
_WEALTH: Dict[str, Wealth] = {}
def register_wealth(name):
return _register_cls(name, dict=_WEALTH)
def get_wealth(name):
return _get_cls(name, dict=_WEALTH)
@register_wealth("mixture")
class Mixture(Wealth):
def __init__(self, config):
super().__init__(config)
self.grid_size = grid_size = config.grid_size
self.wealth = np.ones((grid_size,))
self.wealth_flag = np.ones(grid_size, dtype=bool)
self.v = np.linspace(0.05, 0.95, grid_size)
def update(self, payoff):
raise NotImplementedError
@register_wealth("ons")
class ONS(Wealth):
def __init__(self, config):
super().__init__(config)
self.w = 1.0
self.v = 0
self.a = 1
self.min_v, self.max_v = config.get("min_v", 0), config.get("max_v", 1 / 2)
self.wealth_flag = False
def _update_v(self, payoff):
z = payoff / (1 + self.v * payoff)
self.a += z**2
self.v = max(
self.min_v, min(self.max_v, self.v + 2 / (2 - np.log(3)) * z / self.a)
)
def update(self, payoff):
w = self.w * (1 + self.v * payoff)
if w >= 0 and not self.wealth_flag:
self.w = w
if self.w >= 1 / self.significance_level:
self.rejected = True
self._update_v(payoff)
else:
self.wealth_flag = True
|