Writing an event or a market¶

Events change the simulation while it runs: the built-in events shock the fundamental price, limit order prices, halt trading or inject erroneous orders (see Events). This tutorial writes an event that moves the fundamental price at random at regular intervals, like news, and then shows how to extend the market itself. The complete script is tutorial_custom_event.py.

How the runner uses an event¶

An event is attached to a session by listing the name of its block in the session’s events. For each name listed, the runner:

creates the event. Its name is the name of the block, its session is the session that lists it, and it has its own random number generator prng and the simulator;
calls setup(settings) with the block (after extends is resolved), after all the markets and agents are set up;
calls hook_registration() right after setup. It returns a list of EventHook that tell the simulator which methods of the event to call, and when.

Each hook has a hook_type and is_before, which select the method that is called:

`hook_type`	`is_before=True`	`is_before=False`
`"order"`	`hooked_before_order(simulator, order)`, before a market receives an order. The event can change the `Order` here.	`hooked_after_order(simulator, order_log)`, after a market has received an order.
`"cancel"`	`hooked_before_cancel(simulator, cancel)`	`hooked_after_cancel(simulator, cancel_log)`
`"execution"`	Not allowed.	`hooked_after_execution(simulator, execution_log)`, after each execution.
`"session"`	`hooked_before_session(simulator, session)`, before the first step of a session.	`hooked_after_session(simulator, session)`, after the last step of a session.
`"market"`	`hooked_before_step_for_market(simulator, market)`, before the agents are asked for orders in a step.	`hooked_after_step_for_market(simulator, market)`, at the end of a step.

The hook’s time is the list of steps at which it is active. For a session hook, this is the first step of the session (before) or its last step (after). The steps are counted from the start of the simulation, not from the start of the session. With time=None (the default), the hook is active at every step of every session, not only in the session that lists the event.

A "market" hook also accepts specific_instance (one Market) or specific_class (a market class) to be called only for these markets. Without them, it is called for every market.

The event¶

NewsShock multiplies the fundamental price of its target market by \(e^{x}\) every interval steps of its session, where \(x\) is drawn from a normal distribution with mean 0 and standard deviation jumpScale:

class NewsShock(EventABC):
    """Random news that moves the fundamental price of one market at regular intervals.

    Every ``interval`` steps of its session, the fundamental price of the ``target``
    market is multiplied by :math:`e^{x}`, where :math:`x` is drawn from a normal
    distribution with mean 0 and standard deviation ``jumpScale``.
    """

    target_market: Market
    interval: int
    jump_scale: float
    news_times: List[int]

    def setup(self, settings: Dict[str, Any], *args: Any, **kwargs: Any) -> None:
        """Read the parameters of the event from its block of the config.

        Args:
            settings (Dict[str, Any]): the event's block of the config. It must include
                "target" (the name of a market), "interval" and "jumpScale".
            *args: not used.
            **kwargs: not used.

        """
        for key in ["target", "interval", "jumpScale"]:
            if key not in settings:
                raise ValueError(f"{key} is required for NewsShock")
        self.target_market = self.simulator.name2market[settings["target"]]
        self.interval = int(settings["interval"])
        self.jump_scale = float(settings["jumpScale"])
        self.news_times = []

    def hook_registration(self) -> List[EventHook]:
        """Tell the simulator when to call this event.

        Returns:
            List[EventHook]: a hook before the market step at every news time, and a
            hook after the session of the event.

        """
        start = self.session.session_start_time
        end = start + self.session.iteration_steps
        news_hook = EventHook(
            event=self,
            hook_type="market",
            is_before=True,
            time=list(range(start + self.interval, end, self.interval)),
            specific_instance=self.target_market,
        )
        session_hook = EventHook(
            event=self, hook_type="session", is_before=False, time=[end - 1]
        )
        return [news_hook, session_hook]

    def hooked_before_step_for_market(
        self, simulator: Simulator, market: Market
    ) -> None:
        """Change the fundamental price before the market step.

        Args:
            simulator (Simulator): the simulator.
            market (Market): the target market.

        """
        jump = self.prng.gauss(0.0, self.jump_scale)
        market.change_fundamental_price(scale=math.exp(jump))
        self.news_times.append(market.get_time())

    def hooked_after_session(self, simulator: Simulator, session: Session) -> None:
        """Print a summary at the end of the session of the event.

        Args:
            simulator (Simulator): the simulator.
            session (Session): the session that has just ended.

        """
        print(
            f"{self.name}: {len(self.news_times)} news in {session.name}, "
            f"fundamental price {self.target_market.get_fundamental_price():.2f}"
        )

setup checks and reads the keys of its block. The markets already exist, so it can look up its target with self.simulator.name2market.
hook_registration uses self.session to keep its hooks in its own session: the news hook is active at the steps start + interval, start + 2 * interval, … before the end of the session, and only for the target market. The session hook is active at the last step of the session. With time=None, the session hook would also be called after the warmup session.
change_fundamental_price changes the fundamental price of the current step, before the agents are asked for orders. The change is permanent: the fundamental price stays at the new level, or continues its random walk from it when the market has a fundamentalVolatility.
The event draws its random numbers from self.prng, so the simulation stays determined by its seed.

Using the event¶

The main session lists the event, and the block gives its class and its parameters. The rest is the configuration of Running a simulation:

CONFIG: Dict[str, Any] = {
    "simulation": {
        "markets": ["Market"],
        "agents": ["FCNAgents"],
        "sessions": [
            {
                "sessionName": "warmup",
                "iterationSteps": 100,
                "withOrderPlacement": True,
                "withOrderExecution": False,
                "withPrint": True,
            },
            {
                "sessionName": "main",
                "iterationSteps": 500,
                "withOrderPlacement": True,
                "withOrderExecution": True,
                "withPrint": True,
                "events": ["News"],
            },
        ],
    },
    "News": {
        "class": "NewsShock",
        "target": "Market",
        "interval": 50,
        "jumpScale": 0.02,
    },
    "Market": {"class": "Market", "tickSize": 0.00001, "marketPrice": 300.0},
    "FCNAgents": {
        "class": "FCNAgent",
        "numAgents": 100,
        "markets": ["Market"],
        "assetVolume": 50,
        "cashAmount": 10000,
        "fundamentalWeight": {"expon": [1.0]},
        "chartWeight": {"expon": [0.0]},
        "noiseWeight": {"expon": [1.0]},
        "noiseScale": 0.001,
        "timeWindowSize": [100, 200],
        "orderMargin": [0.0, 0.1],
    },
}

As with agents, register the class with the runner before calling main():

def run_simulation(seed: int) -> SequentialRunner:
    """Register the event class and run the simulation.

    Args:
        seed (int): seed of the random number generator.

    Returns:
        SequentialRunner: the runner after the run.

    """
    runner = SequentialRunner(settings=CONFIG, prng=random.Random(seed))
    runner.class_register(cls=NewsShock)
    runner.main()
    return runner

After the run, runner.simulator.name2event["News"] is the event. Running the complete script prints the summary of the event at the end of main, the two time lines of main(), then the fundamental price before and after each news:

News: 9 news in main, fundamental price 291.61
# INITIALIZATION TIME 0.005436
# EXECUTION TIME 0.1829801
step 150: 300.00 -> 295.17
step 200: 295.17 -> 302.24
step 250: 302.24 -> 301.98
step 300: 301.98 -> 294.94
step 350: 294.94 -> 298.95
step 400: 298.95 -> 298.90
step 450: 298.90 -> 292.69
step 500: 292.69 -> 286.11
step 550: 286.11 -> 291.61

The main session starts at step 100, so the first news is at step 150.

Writing a market¶

Markets are extended in the same way: inherit from Market, call super().setup and read your own keys from the block. samples/market_share uses this market:

class ExtendedMarket(Market):
    def setup(self, settings: Dict[str, Any], *args, **kwargs) -> None:  # type: ignore  # NOQA
        super().setup(settings, *args, **kwargs)
        if "tradeVolume" in settings:
            if not isinstance(settings["tradeVolume"], int):
                raise ValueError("tradeVolume must be int")
            self._executed_volumes = [int(settings["tradeVolume"])]

It sets the traded volume of step 0 from the key tradeVolume. In this sample, MarketShareFCNAgent chooses the market of each order at random, weighted by the volume traded in each market since timeWindowSize steps ago, so tradeVolume sets the initial market shares. The sample registers the class with runner.class_register(cls=ExtendedMarket), and its configuration uses "class": "ExtendedMarket" in the market blocks.

Next, Writing a logger records the results of a simulation in your own format.