-
Notifications
You must be signed in to change notification settings - Fork 83
Expand file tree
/
Copy pathtest_basic.py
More file actions
170 lines (132 loc) · 5.04 KB
/
Copy pathtest_basic.py
File metadata and controls
170 lines (132 loc) · 5.04 KB
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
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
#!/usr/bin/env python3
"""
Basic test script to verify the codebase works with updated dependencies.
"""
import numpy as np
import gymnasium as gym
import gym_ctc_executioner
from ctc_executioner.orderbook import Orderbook
from ctc_executioner.order_side import OrderSide
from ctc_executioner.feature_type import FeatureType
import datetime
def test_environment_creation():
"""Test if we can create and configure the environment."""
print("Testing environment creation...")
# Create artificial orderbook for testing
orderbook = Orderbook()
config = {
"startPrice": 10000.0,
"priceFunction": lambda p0, s, samples: p0
+ 10 * np.sin(2 * np.pi * 10 * (s / samples)),
"levels": 50,
"qtyPosition": 0.1,
"startTime": datetime.datetime.now(),
"duration": datetime.timedelta(seconds=1000), # More states for lookback
"interval": datetime.timedelta(seconds=1),
}
orderbook.createArtificial(config)
print(f"✓ Orderbook created with {len(orderbook.states)} states")
# Create environment
env = gym.make("ctc-executioner-v0")
# Unwrap if needed (Gymnasium wraps environments)
unwrapped_env = env.unwrapped if hasattr(env, "unwrapped") else env
unwrapped_env._configure(
orderbook=orderbook,
side=OrderSide.SELL,
featureType=FeatureType.ORDERS,
lookback=25, # Explicit lookback
bookSize=10,
)
print(f"✓ Environment created")
print(f" Observation space: {env.observation_space.shape}")
print(f" Action space: {env.action_space.n} actions")
return env, unwrapped_env
def test_environment_reset():
"""Test environment reset."""
print("\nTesting environment reset...")
env, unwrapped_env = test_environment_creation()
obs, info = env.reset()
print(f"✓ Reset successful")
print(f" Observation shape: {obs.shape}")
print(f" Info: {info}")
return env, unwrapped_env, obs
def test_environment_step():
"""Test environment step."""
print("\nTesting environment step...")
env, unwrapped_env, obs = test_environment_reset()
# Take a random action
action = env.action_space.sample()
obs_next, reward, terminated, truncated, info = env.step(action)
print(f"✓ Step successful")
print(f" Action: {action}")
print(f" Reward: {reward}")
print(f" Terminated: {terminated}, Truncated: {truncated}")
print(f" Next observation shape: {obs_next.shape}")
return env, unwrapped_env
def test_keras_model():
"""Test if we can create a simple Keras model."""
print("\nTesting Keras model creation...")
from keras.models import Sequential
from keras.layers import Dense, Flatten
from keras import optimizers
# Create a simple model
model = Sequential()
model.add(Flatten(input_shape=(10, 10, 2)))
model.add(Dense(64, activation="relu"))
model.add(Dense(10, activation="linear"))
model.compile(optimizers.Adam(learning_rate=0.001), loss="mse")
print(f"✓ Keras model created")
print(f" Model summary:")
model.summary()
return model
def test_dqn_agent_creation():
"""Test if we can create a DQN agent."""
print("\nTesting DQN agent creation...")
# Import only the class, not the module-level code
import sys
import importlib.util
# Load the module without executing the main code
spec = importlib.util.spec_from_file_location("agent_dqn", "agent_dqn.py")
agent_dqn_module = importlib.util.module_from_spec(spec)
# Don't execute the module to avoid file loading
# Instead, we'll create the agent class directly
from agent_dqn import AgentDQN
env, unwrapped_env = test_environment_creation()
agent = AgentDQN(env)
print(f"✓ DQN agent created")
print(f" Action size: {agent.action_size}")
print(f" Model built successfully")
return agent, env, unwrapped_env
def test_dqn_agent_prediction():
"""Test if DQN agent can make predictions."""
print("\nTesting DQN agent prediction...")
agent, env, unwrapped_env = test_dqn_agent_creation()
obs, _ = env.reset()
# Add batch dimension
obs_batch = np.expand_dims(obs, axis=0)
action = agent.act(obs_batch)
print(f"✓ Agent prediction successful")
print(f" Selected action: {action}")
return agent, env, unwrapped_env
if __name__ == "__main__":
print("=" * 60)
print("Basic Functionality Test")
print("=" * 60)
try:
# Run tests (skip DQN tests that require file loading)
test_environment_creation()
test_environment_reset()
test_environment_step()
test_keras_model()
# Skip DQN tests for now as they require data files
# test_dqn_agent_creation()
# test_dqn_agent_prediction()
print("\n" + "=" * 60)
print("✓ Core functionality tests passed!")
print(" (DQN agent tests skipped - require data files)")
print("=" * 60)
except Exception as e:
print(f"\n✗ Test failed with error: {e}")
import traceback
traceback.print_exc()
exit(1)