Function run_basic_simulation

market_simulation/examples/run_simulation.py:19–69

Run a basic market simulation with a noise agent. This function creates a simplified market environment with a single agent that generates random order flow. The simulation runs for a specified time period and then visualizes the resulting price trajectory. Args: seed: Rand

(seed: int = 0)

def run_basic_simulation(seed: int = 0) -> None:
    """Run a basic market simulation with a noise agent.

    This function creates a simplified market environment with a single agent
    that generates random order flow. The simulation runs for a specified time period
    and then visualizes the resulting price trajectory.

    Args:
        seed: Random seed for reproducible simulation results

    Returns:
        None
    """
    # Setup simulation parameters
    symbols = ["000000"]
    start_time = Timestamp("2024-01-01 09:30:00")
    end_time = Timestamp("2024-01-01 10:30:00")

    # Create exchange environment
    exchange_config = create_exchange_config_without_call_auction(
        market_open=start_time,
        market_close=end_time,
        symbols=symbols,
    )
    exchange = Exchange(exchange_config)

    # Initialize noise agent for order generation
    agent = NoiseAgent(
        symbol=symbols[0],
        init_price=100000,
        interval_seconds=1,
        start_time=start_time,
        end_time=end_time,
        seed=seed,
    )

    # Configure simulation environment
    exchange.register_state(TradeInfoState())
    env = Env(exchange=exchange, description="Noise agent simulation")
    env.register_agent(agent)
    env.push_events(create_exchange_events(exchange_config))

    # Run simulation
    for observation in env.env():
        action = observation.agent.get_action(observation)
        env.step(action)

    # Extract and visualize results
    trade_infos: list[TradeInfo] = extract_trade_information(exchange, symbols[0], start_time, end_time)
    logging.info(f"Collected {len(trade_infos)} trade information records.")
    visualize_price_trajectory(trade_infos, Path("tmp/price_curves.png"))


def extract_trade_information(exchange: Exchange, symbol: str, start_time: Timestamp, end_time: Timestamp) -> list[TradeInfo]: