Merge pull request #81 from masanorihirano/takata/samples_price_limit…

…_rule

[samples] PriceLimit

docs/source/reference/events.rst

@@ -9,3 +9,4 @@ Events
     events.EventABC
     events.EventHook
     events.FundamentalPriceShock
+    events.PriceLimitRule

docs/source/user_guide/config.rst

@@ -36,6 +36,12 @@ Json config
             "shockTimeLength": int (Optional, default: 1),
             "enabled": bool (Optional, default: True)
         },
+        "PriceLimitRule": {
+            "class": "PriceLimitRule",
+            "targetMarkets": ["Market"],
+            "triggerChangeRate": float required,
+            "enabled": bool (Optional, default: True)
+        },
         "Market": {
             "extends": string (Optional),
             "class": string,

pams/events/__init__.py

@@ -1,3 +1,4 @@
 from .base import EventABC
 from .base import EventHook
 from .fundamental_price_shock import FundamentalPriceShock
+from .price_limit_rule import PriceLimitRule

pams/events/price_limit_rule.py

@@ -0,0 +1,122 @@
+import random
+import warnings
+from typing import Any
+from typing import Dict
+from typing import List
+from typing import Optional
+
+from ..market import Market
+from ..order import Order
+from ..session import Session
+from ..simulator import Simulator
+from .base import EventABC
+from .base import EventHook
+
+
+class PriceLimitRule(EventABC):
+    """This limits the price range.
+
+    The order having the price that is out of the price range, the price is overridden to the edge of range.
+    This event is only called via :func:`hooked_before_order` at designated step.
+    """
+
+    reference_price: float
+    trigger_change_rate: float
+
+    def __init__(
+        self,
+        event_id: int,
+        prng: random.Random,
+        session: Session,
+        simulator: Simulator,
+        name: str,
+    ) -> None:
+        super().__init__(
+            event_id=event_id,
+            prng=prng,
+            session=session,
+            simulator=simulator,
+            name=name,
+        )
+        self.target_markets: Dict[str, Market] = {}
+        self.is_enabled: bool = True
+        self.activation_count: int = 0
+
+    def setup(self, settings: Dict[str, Any], *args, **kwargs) -> None:  # type: ignore  # NOQA
+        """event setup. Usually be called from simulator/runner automatically.
+
+        Args:
+            settings (Dict[str, Any]): agent configuration. Usually, automatically set from json config of simulator.
+                                       This must include the parameters "targetMarkets" and "triggerChangeRate".
+                                       This can include the parameters "enabled".
+                                       The parameter "referenceMarket" is obsoleted.
+
+        Returns:
+            None
+        """
+        if "referenceMarket" in settings:
+            warnings.warn("referenceMarket is obsoleted")
+        if "targetMarkets" not in settings:
+            raise ValueError("targetMarkets is required for PriceLimitRule")
+        if not isinstance(settings["targetMarkets"], list):
+            raise ValueError("targetMarkets must be list")
+        for market_name in settings["targetMarkets"]:
+            if not isinstance(market_name, str):
+                raise ValueError("constituent of targetMarkets have to be string")
+            if market_name not in self.simulator.name2market:
+                raise ValueError(f"{market_name} does not exist")
+            market: Market = self.simulator.name2market[market_name]
+            self.target_markets[market_name] = market
+        if "triggerChangeRate" not in settings:
+            raise ValueError("triggerChangeRate is required for PriceLimitRule")
+        if not isinstance(settings["triggerChangeRate"], float):
+            raise ValueError("triggerChangeRate have to be float.")
+        self.trigger_change_rate = settings["triggerChangeRate"]
+        if "enabled" in settings:
+            self.is_enabled = settings["enabled"]
+
+    def hook_registration(self) -> List[EventHook]:
+        if self.is_enabled:
+            event_hook = EventHook(
+                event=self, hook_type="order", is_before=True, time=None
+            )
+            return [event_hook]
+        else:
+            return []
+
+    def get_limited_price(self, order: Order, market: Market) -> Optional[float]:
+        """Calculate the limited price for an order.
+
+        Args:
+            order (Order): order whose price is calculated
+            market (Market): market that order belongs to
+
+        Returns:
+            Optional[float]: price after price limit. If the input order is market order, the return become None (market order).
+        """
+        self.reference_price = market.get_market_price(0)
+        if market not in self.target_markets.values():
+            raise AssertionError
+        if order.price is None:
+            return order.price
+        order_price: float = order.price
+        price_change: float = order_price - self.reference_price
+        threshold_change: float = self.reference_price * self.trigger_change_rate
+        if abs(price_change) >= abs(threshold_change):
+            max_price: float = self.reference_price * (1 + self.trigger_change_rate)
+            min_price: float = self.reference_price * (1 - self.trigger_change_rate)
+            limited_price: float = min(max(order_price, min_price), max_price)
+            return limited_price
+        return order_price
+
+    def hooked_before_order(self, simulator: Simulator, order: Order) -> None:
+        new_price: Optional[float] = self.get_limited_price(
+            order, simulator.id2market[order.market_id]
+        )
+        if order.price != new_price:
+            self.activation_count += 1
+        order.price = new_price
+
+
+PriceLimitRule.hook_registration.__doc__ = EventABC.hook_registration.__doc__
+PriceLimitRule.hooked_before_order.__doc__ = EventABC.hooked_before_order.__doc__

samples/price_limit/config.json

@@ -0,0 +1,51 @@
+{
+	"simulation": {
+		"markets": ["Market"],
+		"agents": ["FCNAgents"],
+		"sessions": [
+			{	"sessionName": 0,
+				"iterationSteps": 100,
+				"withOrderPlacement": true,
+				"withOrderExecution": false,
+				"withPrint": true
+			},
+			{	"sessionName": 1,
+				"iterationSteps": 500,
+				"withOrderPlacement": true,
+				"withOrderExecution": true,
+				"withPrint": true,
+				"events": ["PriceLimitRule"]
+			}
+		]
+	},
+
+	"PriceLimitRule": {
+		"class": "PriceLimitRule",
+		"targetMarkets": ["Market"],
+		"triggerChangeRate": 0.05,
+		"enabled": true
+	},
+
+	"Market": {
+		"class": "Market",
+		"tickSize": 0.00001,
+		"marketPrice": 300.0,
+		"outstandingShares": 25000
+	},
+
+	"FCNAgents": {
+		"class": "FCNAgent",
+		"numAgents": 100,
+
+		"markets": ["Market"],
+		"assetVolume": 50,
+		"cashAmount": 10000,
+
+		"fundamentalWeight": {"expon": [0.2]},
+		"chartWeight": {"expon": [0.0]},
+		"noiseWeight": {"expon": [1.0]},
+		"noiseScale": 0.001,
+		"timeWindowSize": [100, 200],
+		"orderMargin": [0.0, 0.1]
+	}
+}

samples/price_limit/main.py

@@ -0,0 +1,30 @@
+import argparse
+import random
+from typing import Optional
+
+from pams.logs.market_step_loggers import MarketStepPrintLogger
+from pams.runners.sequential import SequentialRunner
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--config", "-c", type=str, required=True, help="config.json file"
+    )
+    parser.add_argument(
+        "--seed", "-s", type=int, default=None, help="simulation random seed"
+    )
+    args = parser.parse_args()
+    config: str = args.config
+    seed: Optional[int] = args.seed
+
+    runner = SequentialRunner(
+        settings=config,
+        prng=random.Random(seed) if seed is not None else None,
+        logger=MarketStepPrintLogger(),
+    )
+    runner.main()
+
+
+if __name__ == "__main__":
+    main()