Add files via upload

config_ma2c_real.ini

@@ -0,0 +1,47 @@
+[MODEL_CONFIG]
+rmsp_alpha = 0.99
+rmsp_epsilon = 1e-5
+max_grad_norm = 40
+gamma = 0.99
+lr_init = 5e-4
+lr_decay = constant
+entropy_coef_init = 0.01
+entropy_coef_min = 0.01
+entropy_decay = constant
+entropy_ratio = 0.5
+value_coef = 0.5
+num_fw = 128
+num_ft = 32
+num_lstm = 64
+num_fp = 64
+batch_size = 10
+reward_norm = 1.0
+reward_clip = 2.0
+
+[TRAIN_CONFIG]
+total_step = 2e6
+test_interval = 2e4
+log_interval = 1e4
+
+[ENV_CONFIG]
+clip_wave = 2.0
+clip_wait = 2.0
+; agent is greedy, iqll, iqld, ia2c, ma2c, a2c.
+agent = ma2c
+; coop discount is used to discount the neighbors' impact
+coop_gamma = 0.9
+data_path = ./real_net/data/
+price_data = ./real_net/price_data/
+key = EURUSD_lastmonth
+window_size = 10
+balance = 995
+; the normailization is based on typical values in sim
+norm_wave = 5.0
+norm_wait = 30.0
+coef_wait = 0
+; objective is chosen from queue, wait, hybrid
+objective = queue
+scenario = real_net
+seed = 42
+test_seeds = 10000,20000,30000
+yellow_interval_sec = 2

evaluate.sh

@@ -0,0 +1 @@
+python3 main.py --base-dir real_net evaluate --agents ma2c --evaluation-policy-type deterministic

get_bars.py

@@ -0,0 +1,40 @@
+import pytz
+from datetime import datetime
+import MetaTrader5 as mt5
+import pandas as pd
+pd.set_option('display.max_columns', 500)  # number of columns to be displayed
+pd.set_option('display.width', 1500)      # max table width to display
+# import pytz module for working with time zone
+
+# establish connection to MetaTrader 5 terminal
+mt5.initialize()
+# file name to export to csv
+file_name = 'EURUSD_H4_20220103_20220203.csv'
+# set time zone to UTC
+timezone = pytz.timezone("Etc/GMT-2")
+# create 'datetime' objects in UTC time zone to avoid the implementation of a local time zone offset
+utc_from = datetime(2022, 1, 3, tzinfo=timezone)
+utc_to = datetime(2022, 2, 3, tzinfo=timezone)
+# get bars from EURUSD H4 within the interval of 2021.05.03 00:00 - 2022.01.03 00:00 in GMT-2 time zone
+rates = mt5.copy_rates_range("EURUSD", mt5.TIMEFRAME_H4, utc_from, utc_to)
+
+# shut down connection to the MetaTrader 5 terminal
+mt5.shutdown()
+
+# display each element of obtained data in a new line
+print("Display obtained data 'as is'")
+counter = 0
+for rate in rates:
+    counter += 1
+    if counter <= 10:
+        print(rate)
+
+# create DataFrame out of the obtained data
+rates_frame = pd.DataFrame(rates)
+# convert time in seconds into the 'datetime' format
+rates_frame['time'] = pd.to_datetime(rates_frame['time'], unit='s')
+
+# display data
+print("\nDisplay dataframe with data")
+print(rates_frame.head(10))
+rates_frame.to_csv(file_name, encoding='utf-8', index=False)

infer.py

@@ -0,0 +1,207 @@
+import MetaTrader5 as mt5
+import pandas as pd
+import numpy as np
+import math
+import argparse
+import configparser
+from envs.real_net_env import RealNetEnv, RealNetController
+from envs.functions import getState, formatPrice
+from agents.models import MA2C
+from utils import Predictor
+
+
+SYMBOL = "EURUSD"
+DEVIATION = 20
+TIMEFRAME = mt5.TIMEFRAME_H4
+VOLUME = 0.03
+PERIOD = 11
+
+
+def parse_args():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--config-dir', type=str, required=False,
+                        default='config_ma2c_real.ini', help="inference config dir")
+    parser.add_argument('--port', type=int, required=False,
+                        default=0, help="running port")
+    parser.add_argument('--policy-type', type=str, required=False, default='default',
+                        help="inference policy type in evaluation: default, stochastic, or deterministic")
+    parser.add_argument('--position-type', type=dict, required=False,
+                        default={'long': 1, 'short': -1}, help="types of position")
+    args = parser.parse_args()
+    return args
+
+
+def market_order(symbol, volume, order_type):
+    tick = mt5.symbol_info_tick(symbol)
+
+    order_dict = {'long': 0, 'short': 1}
+    price_dict = {'long': tick.ask, 'short': tick.bid}
+
+    request = {
+        "action": mt5.TRADE_ACTION_DEAL,
+        "symbol": symbol,
+        "volume": volume,
+        "type": order_dict[order_type],
+        "price": price_dict[order_type],
+        "deviation": DEVIATION,
+        "magic": 100,
+        "comment": "python market order",
+        "type_time": mt5.ORDER_TIME_GTC,
+        "type_filling": mt5.ORDER_FILLING_IOC,
+    }
+
+    order_result = mt5.order_send(request)
+    print(order_result)
+
+    return order_result
+
+
+# function to close an order base don ticket id
+def close_order(ticket):
+    positions = mt5.positions_get()
+
+    for pos in positions:
+        tick = mt5.symbol_info_tick(pos.symbol)
+        # 0 represents buy, 1 represents sell - inverting order_type to close the position
+        type_dict = {0: 1, 1: 0}
+        price_dict = {0: tick.ask, 1: tick.bid}
+
+        if pos.ticket == ticket:
+            request = {
+                "action": mt5.TRADE_ACTION_DEAL,
+                "position": pos.ticket,
+                "symbol": pos.symbol,
+                "volume": pos.volume,
+                "type": type_dict[pos.type],
+                "price": price_dict[pos.type],
+                "deviation": DEVIATION,
+                "magic": 100,
+                "comment": "python close order",
+                "type_time": mt5.ORDER_TIME_GTC,
+                "type_filling": mt5.ORDER_FILLING_IOC,
+            }
+            order_result = mt5.order_send(request)
+            print(order_result)
+
+            return order_result
+
+    return 'Ticket does not exist'
+
+
+def sigmoid(x):
+    return 1 / (1 + math.exp(-x))
+
+
+def _norm_clip_state(x, norm, clip=-1):
+    x = x / norm
+    return x if clip < 0 else np.clip(x, 0, clip)
+
+
+def getState(symbol, timeframe, period, index=1000):
+    bars = mt5.copy_rates_from_pos(symbol, timeframe, 1, period)
+    bars_df = pd.DataFrame(bars)
+    vec = bars_df.close.tolist()
+    vec = [x*index for x in vec]
+    res = []
+    for i in range(period - 1):
+        res.append(sigmoid(vec[i + 1] - vec[i]))
+
+    return np.array(res)
+
+
+def init_env(config, port=1, naive_policy=False):
+    if not naive_policy:
+        return RealNetEnv(config, port=port)
+    else:
+        env = RealNetEnv(config, port=port)
+        policy = RealNetController(env.node_names, env.nodes)
+        return env, policy
+
+
+def data_preprocessing(cur_state, norm, clip, agents):
+    # hard code the state ordering as wave, wait, fp
+    state = []
+    # measure the most recent state
+    norm_cur_state = _norm_clip_state(cur_state, norm, clip)
+    # get the appropriate state vectors
+    for _ in agents:
+        # wave is required in state
+        cur_state = [norm_cur_state]
+        state.append(np.concatenate(cur_state))
+
+    return state
+
+
+def main(args):
+    config_dir = args.config_dir
+    port = args.port
+    policy_type = args.policy_type
+    agent_type = args.position_type
+
+    # initialize start value
+    inventory = {}
+    open_ticket = {}
+    for agent in [*agent_type]:
+        inventory[agent] = []
+        open_ticket[agent] = []
+    total_profit = 0
+    balance_list = []
+    pre_state = np.array([])
+
+    # load config file for env
+    config = configparser.ConfigParser()
+    config.read(config_dir)
+    cur_balance = config['ENV_CONFIG'].getint('balance')
+    norm = config['ENV_CONFIG'].getfloat('norm_wave')
+    clip = config['ENV_CONFIG'].getfloat('clip_wave')
+
+    # init env
+    env = init_env(config['ENV_CONFIG'], port)
+    # load model for agent
+    # init centralized or multi agent
+    model = MA2C(env.n_s_ls, env.n_a_ls, env.n_w_ls,
+                 env.n_f_ls, 0, config['MODEL_CONFIG'])
+    model.load('weights/')
+    model.reset()
+    # collect evaluation data
+    predictor = Predictor(env, model, policy_type=policy_type)
+    # init mt5
+    mt5.initialize()
+
+    while True:
+        cur_state = getState(symbol=SYMBOL, timeframe=TIMEFRAME,
+                             period=PERIOD)
+        if not np.array_equal(cur_state, pre_state):
+            state = data_preprocessing(cur_state, norm, clip, [*agent_type])
+            action = predictor.run(state)
+            print('---ACTION--- :', action)
+            tick = mt5.symbol_info_tick(SYMBOL)
+            price_dict = {'long': tick.ask, 'short': tick.bid}
+
+            for agent, a in zip([*agent_type], list(action)):
+                if a == 1:
+                    market_order(SYMBOL, VOLUME, agent)
+                    inventory[agent].append(price_dict[agent])
+                    open_ticket[agent].append(mt5.positions_get()[-1].ticket)
+
+                elif a == 2 and len(inventory[agent]) > 0:  
+                    close_order(open_ticket[agent].pop(0))
+                    order_price = inventory[agent].pop(0)
+                    profit = (price_dict[agent] - order_price) * \
+                        agent_type[agent] * VOLUME * 100000
+                    total_profit += profit
+                    cur_balance += profit
+                    balance_list.append(round(cur_balance, 2))
+
+            print("--------------------------------")
+            print("Total Profit: " + formatPrice(total_profit))
+            print('Curent Balance: ' + formatPrice(cur_balance))
+            print('Balance List', balance_list)
+            print("--------------------------------")
+
+        pre_state = cur_state
+
+
+if __name__ == '__main__':
+    args = parse_args()
+    main(args)

main.py

@@ -0,0 +1,160 @@
+"""
+Main function for training and evaluating agents in traffic envs
+@author: Tianshu Chu
+run command: 
+1. Train: python main.py --base-dir real_net/ma2c train --config-dir config/config_ma2c_real.ini --test-mode no_test
+2. Visualize: python main.py --base-dir real_net evaluate --agents ma2c 
+"""
+
+import argparse
+import configparser
+import logging
+import tensorflow.compat.v1 as tf
+import threading
+from envs.real_net_env import RealNetEnv, RealNetController
+from agents.models import MA2C
+from utils import (Counter, Trainer, Tester, Evaluator,
+                   check_dir, copy_file, find_file,
+                   init_dir, init_log, init_test_flag)
+
+
+def parse_args():
+    default_base_dir = '/Users/tchu/Documents/rl_test/signal_control_results/eval_sep2019/large_grid'
+    default_config_dir = './config/config_test_large.ini'
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--base-dir', type=str, required=False,
+                        default=default_base_dir, help="experiment base dir")
+    subparsers = parser.add_subparsers(dest='option', help="train or evaluate")
+    sp = subparsers.add_parser(
+        'train', help='train a single agent under base dir')
+    sp.add_argument('--test-mode', type=str, required=False,
+                    default='no_test',
+                    help="test mode during training",
+                    choices=['no_test', 'in_train_test', 'after_train_test', 'all_test'])
+    sp.add_argument('--config-dir', type=str, required=False,
+                    default=default_config_dir, help="experiment config path")
+    sp = subparsers.add_parser(
+        'evaluate', help="evaluate and compare agents under base dir")
+    sp.add_argument('--agents', type=str, required=False,
+                    default='naive', help="agent folder names for evaluation, split by ,")
+    sp.add_argument('--evaluation-policy-type', type=str, required=False, default='default',
+                    help="inference policy type in evaluation: default, stochastic, or deterministic")
+    args = parser.parse_args()
+    if not args.option:
+        parser.print_help()
+        exit(1)
+    return args
+
+
+def init_env(config, port=1, naive_policy=False):
+    if not naive_policy:
+        return RealNetEnv(config, port=port)
+    else:
+        env = RealNetEnv(config, port=port)
+        policy = RealNetController(env.node_names, env.nodes)
+        return env, policy
+
+
+def train(args):
+    base_dir = args.base_dir
+    dirs = init_dir(base_dir)
+    init_log(dirs['log'])
+    config_dir = args.config_dir
+    copy_file(config_dir, dirs['data'])
+    config = configparser.ConfigParser()
+    config.read(config_dir)
+    in_test, post_test = init_test_flag(args.test_mode)
+
+    # init env
+    env = init_env(config['ENV_CONFIG'])
+    logging.info('Training: s dim: %d, a dim %d, s dim ls: %r, a dim ls: %r' %
+                 (env.n_s, env.n_a, env.n_s_ls, env.n_a_ls))
+
+    # init step counter
+    total_step = int(config.getfloat('TRAIN_CONFIG', 'total_step'))
+    test_step = int(config.getfloat('TRAIN_CONFIG', 'test_interval'))
+    log_step = int(config.getfloat('TRAIN_CONFIG', 'log_interval'))
+    global_counter = Counter(total_step, test_step, log_step)
+
+    # init centralized or multi agent
+    seed = config.getint('ENV_CONFIG', 'seed')
+    model = MA2C(env.n_s_ls, env.n_a_ls, env.n_w_ls, env.n_f_ls, total_step,
+                 config['MODEL_CONFIG'], seed=seed)
+
+    # disable multi-threading for safe SUMO implementation
+    summary_writer = tf.summary.FileWriter(dirs['log'])
+    trainer = Trainer(env, model, global_counter,
+                      summary_writer, in_test, output_path=dirs['data'])
+    trainer.run()
+    # post-training test
+    if post_test:
+        tester = Tester(env, model, global_counter,
+                        summary_writer, dirs['data'])
+        tester.run_offline(dirs['data'])
+
+    # save model
+    final_step = global_counter.cur_step
+    logging.info('Training: save final model at step %d ...' % final_step)
+    model.save(dirs['model'], final_step)
+
+
+def evaluate_fn(agent_dir, output_dir, port, policy_type):
+    agent = agent_dir.split('/')[-1]
+    if not check_dir(agent_dir):
+        logging.error('Evaluation: %s does not exist!' % agent)
+        return
+    # load config file for env
+    config_dir = find_file(agent_dir + '/data/')
+    if not config_dir:
+        return
+    config = configparser.ConfigParser()
+    config.read(config_dir)
+
+    # init env
+    env = init_env(config['ENV_CONFIG'], port)
+    logging.info('Evaluation: s dim: %d, a dim %d, s dim ls: %r, a dim ls: %r' %
+                 (env.n_s, env.n_a, env.n_s_ls, env.n_a_ls))
+
+    # load model for agent
+    # init centralized or multi agent
+    model = MA2C(env.n_s_ls, env.n_a_ls, env.n_w_ls,
+                 env.n_f_ls, 0, config['MODEL_CONFIG'])
+    if not model.load(agent_dir + '/model/'):
+        return
+    print('agent', agent)
+    print('env.agent', env.agent)
+    env.agent = agent
+    # collect evaluation data
+    evaluator = Evaluator(env, model, output_dir, policy_type=policy_type)
+    evaluator.run()
+
+
+def evaluate(args):
+    base_dir = args.base_dir
+    dirs = init_dir(base_dir, pathes=['eva_data', 'eva_log'])
+    init_log(dirs['eva_log'])
+    agents = args.agents.split(',')
+    print('agents', agents)
+    # enforce the same evaluation seeds across agents
+    policy_type = args.evaluation_policy_type
+    logging.info('Evaluation: policy type: %s' %
+                 (policy_type))
+
+    threads = []
+    for i, agent in enumerate(agents):
+        print('agent', agent)
+        agent_dir = base_dir + '/' + agent
+        thread = threading.Thread(target=evaluate_fn,
+                                  args=(agent_dir, dirs['eva_data'], i, policy_type))
+        thread.start()
+        threads.append(thread)
+    for thread in threads:
+        thread.join()
+
+
+if __name__ == '__main__':
+    args = parse_args()
+    if args.option == 'train':
+        train(args)
+    else:
+        evaluate(args)

train.sh

@@ -0,0 +1 @@
+python3 main.py --base-dir real_net/ma2c train --config-dir config/config_ma2c_real.ini --test-mode no_test

utils.py

@@ -0,0 +1,383 @@
+import itertools
+import logging
+import numpy as np
+import tensorflow
+import tensorflow.compat.v1 as tf
+import time
+import os
+import pandas as pd
+import subprocess
+
+
+def check_dir(cur_dir):
+    if not os.path.exists(cur_dir):
+        return False
+    return True
+
+
+def copy_file(src_dir, tar_dir):
+    cmd = 'cp %s %s' % (src_dir, tar_dir)
+    subprocess.check_call(cmd, shell=True)
+
+
+def find_file(cur_dir, suffix='.ini'):
+    for file in os.listdir(cur_dir):
+        if file.endswith(suffix):
+            return cur_dir + '/' + file
+    logging.error('Cannot find %s file' % suffix)
+    return None
+
+
+def init_dir(base_dir, pathes=['log', 'data', 'model']):
+    if not os.path.exists(base_dir):
+        os.mkdir(base_dir)
+    dirs = {}
+    for path in pathes:
+        cur_dir = base_dir + '/%s/' % path
+        if not os.path.exists(cur_dir):
+            os.mkdir(cur_dir)
+        dirs[path] = cur_dir
+    return dirs
+
+
+def init_log(log_dir):
+    logging.basicConfig(format='%(asctime)s [%(levelname)s] %(message)s',
+                        level=logging.INFO,
+                        handlers=[
+                            logging.FileHandler('%s/%d.log' %
+                                                (log_dir, time.time())),
+                            logging.StreamHandler()
+                        ])
+
+
+def init_test_flag(test_mode):
+    if test_mode == 'no_test':
+        return False, False
+    if test_mode == 'in_train_test':
+        return True, False
+    if test_mode == 'after_train_test':
+        return False, True
+    if test_mode == 'all_test':
+        return True, True
+    return False, False
+
+
+class Counter:
+    def __init__(self, total_step, test_step, log_step):
+        self.counter = itertools.count(1)
+        self.cur_step = 0
+        self.cur_test_step = 0
+        self.total_step = total_step
+        self.test_step = test_step
+        self.log_step = log_step
+        self.stop = False
+        # self.init_test = True
+
+    def next(self):
+        self.cur_step = next(self.counter)
+        return self.cur_step
+
+    def should_test(self):
+        test = False
+        if (self.cur_step - self.cur_test_step) >= self.test_step:
+            test = True
+            self.cur_test_step = self.cur_step
+        return test
+
+    def should_log(self):
+        return (self.cur_step % self.log_step == 0)
+
+    def should_stop(self):
+        if self.cur_step >= self.total_step:
+            return True
+        return self.stop
+
+
+class Trainer():
+    def __init__(self, env, model, global_counter, summary_writer, run_test, output_path=None):
+        self.cur_step = 0
+        self.global_counter = global_counter
+        self.env = env
+        self.agent = self.env.agent
+        self.model = model
+        self.sess = self.model.sess
+        self.n_step = self.model.n_step
+        self.summary_writer = summary_writer
+        self.run_test = run_test
+        self.data = []
+        self.output_path = output_path
+        if run_test:
+            self.test_num = self.env.test_num
+            logging.info('Testing: total test num: %d' % self.test_num)
+        self._init_summary()
+
+    def _init_summary(self):
+        self.train_reward = tf.placeholder(tensorflow.float32, [])
+        self.train_summary = tf.summary.scalar(
+            'train_reward', self.train_reward)
+        self.test_reward = tf.placeholder(tensorflow.float32, [])
+        self.test_summary = tf.summary.scalar('test_reward', self.test_reward)
+
+    def _add_summary(self, reward, global_step, is_train=True):
+        if is_train:
+            summ = self.sess.run(self.train_summary, {
+                                 self.train_reward: reward})
+        else:
+            summ = self.sess.run(self.test_summary, {self.test_reward: reward})
+        self.summary_writer.add_summary(summ, global_step=global_step)
+
+    def explore(self, prev_ob, prev_done):
+        ob = prev_ob
+        done = prev_done
+        rewards = []
+        for _ in range(self.n_step):
+            policy, value = self.model.forward(ob, done)
+            # need to update fingerprint before calling step
+            self.env.update_fingerprint(policy)
+            action = []
+            for pi in policy:
+                action.append(np.random.choice(
+                    np.arange(len(pi)), p=pi))
+
+            next_ob, reward, done, global_reward = self.env.step(action)
+            rewards.append(global_reward)
+            global_step = self.global_counter.next()
+            self.cur_step += 1
+            if self.agent.endswith('a2c'):
+                self.model.add_transition(ob, action, reward, value, done)
+            else:
+                self.model.add_transition(ob, action, reward, next_ob, done)
+            # logging
+            if self.global_counter.should_log():
+                logging.info('''Training: global step %d, episode step %d,
+                                   ob: %s, a: %s, pi: %s, r: %.2f, train r: %.2f, done: %r''' %
+                             (global_step, self.cur_step,
+                              str(ob), str(action), str(policy), global_reward, np.mean(reward), done))
+
+            if done:
+                break
+            ob = next_ob
+        if self.agent.endswith('a2c'):
+            if done:
+                R = 0 if self.agent == 'a2c' else [0] * self.model.n_agent
+            else:
+                R = self.model.forward(ob, False, 'v')
+        else:
+            R = 0
+        return ob, done, R, rewards
+
+    def inference(self, ob, policy_type='default'):
+        # note this done is pre-decision to reset LSTM states!
+        done = False
+        # self.model.reset()
+        # policy-based on-poicy learning
+        policy = self.model.forward(ob, done, 'p')
+        self.env.update_fingerprint(policy)
+        action = []
+        for pi in policy:
+            if policy_type != 'deterministic':
+                action.append(np.random.choice(
+                    np.arange(len(pi)), p=pi))
+            else:
+                action.append(np.argmax(np.array(pi)))
+
+        return action
+
+    def perform(self, policy_type='default'):
+        ob = self.env.reset()
+        # note this done is pre-decision to reset LSTM states!
+        done = True
+        self.model.reset()
+        rewards = []
+        while True:
+            # policy-based on-poicy learning
+            policy = self.model.forward(ob, done, 'p')
+            self.env.update_fingerprint(policy)
+            action = []
+            for pi in policy:
+                if policy_type != 'deterministic':
+                    action.append(np.random.choice(
+                        np.arange(len(pi)), p=pi))
+                else:
+                    action.append(np.argmax(np.array(pi)))
+
+            next_ob, reward, done, global_reward = self.env.step(action)
+            rewards.append(global_reward)
+            if done:
+                break
+            ob = next_ob
+        mean_reward = np.mean(np.array(rewards))
+        std_reward = np.std(np.array(rewards))
+        return mean_reward, std_reward
+
+    def run_thread(self, coord):
+        '''Multi-threading is disabled'''
+        ob = self.env.reset()
+        done = False
+        cum_reward = 0
+        while not coord.should_stop():
+            ob, done, R, cum_reward = self.explore(ob, done, cum_reward)
+            global_step = self.global_counter.cur_step
+            if self.agent.endswith('a2c'):
+                self.model.backward(R, self.summary_writer, global_step)
+            else:
+                self.model.backward(self.summary_writer, global_step)
+            self.summary_writer.flush()
+            if (self.global_counter.should_stop()) and (not coord.should_stop()):
+                self.env.terminate()
+                coord.request_stop()
+                logging.info('Training: stop condition reached!')
+                return
+
+    def run(self):
+        while not self.global_counter.should_stop():
+            # test
+            if self.run_test and self.global_counter.should_test():
+                rewards = []
+                global_step = self.global_counter.cur_step
+                self.env.train_mode = False
+                for test_ind in range(self.test_num):
+                    mean_reward, std_reward = self.perform(test_ind)
+                    self.env.terminate()
+                    rewards.append(mean_reward)
+                    log = {'agent': self.agent,
+                           'step': global_step,
+                           'test_id': test_ind,
+                           'avg_reward': mean_reward,
+                           'std_reward': std_reward}
+                    self.data.append(log)
+                avg_reward = np.mean(np.array(rewards))
+                self._add_summary(avg_reward, global_step, is_train=False)
+                logging.info('Testing: global step %d, avg R: %.2f' %
+                             (global_step, avg_reward))
+            # train
+            self.env.train_mode = True
+            ob = self.env.reset()
+            # note this done is pre-decision to reset LSTM states!
+            done = True
+            self.model.reset()
+            self.cur_step = 0
+            rewards = []
+            while True:
+                ob, done, R, cur_rewards = self.explore(ob, done)
+                rewards += cur_rewards
+                global_step = self.global_counter.cur_step
+                if self.agent.endswith('a2c'):
+                    self.model.backward(R, self.summary_writer, global_step)
+                else:
+                    self.model.backward(self.summary_writer, global_step)
+                # termination
+                if done:
+                    # self.env.terminate()
+                    break
+            rewards = np.array(rewards)
+            mean_reward = np.mean(rewards)
+            std_reward = np.std(rewards)
+            log = {'agent': self.agent,
+                   'step': global_step,
+                   'test_id': -1,
+                   'avg_reward': mean_reward,
+                   'std_reward': std_reward}
+            self.data.append(log)
+            self._add_summary(mean_reward, global_step)
+            self.summary_writer.flush()
+        df = pd.DataFrame(self.data)
+        df.to_csv(self.output_path + 'train_reward.csv')
+
+
+class Tester(Trainer):
+    def __init__(self, env, model, global_counter, summary_writer, output_path):
+        super().__init__(env, model, global_counter, summary_writer)
+        self.env.train_mode = False
+        self.test_num = self.env.test_num
+        self.output_path = output_path
+        self.data = []
+        logging.info('Testing: total test num: %d' % self.test_num)
+
+    def _init_summary(self):
+        self.reward = tf.placeholder(tensorflow.float32, [])
+        self.summary = tf.summary.scalar('test_reward', self.reward)
+
+    def run_offline(self):
+        # enable traffic measurments for offline test
+        is_record = True
+        record_stats = False
+        self.env.cur_episode = 0
+        self.env.init_data(is_record, record_stats, self.output_path)
+        rewards = []
+        for test_ind in range(self.test_num):
+            rewards.append(self.perform(test_ind))
+            self.env.terminate()
+            time.sleep(2)
+            self.env.collect_tripinfo()
+        avg_reward = np.mean(np.array(rewards))
+        logging.info('Offline testing: avg R: %.2f' % avg_reward)
+        self.env.output_data()
+
+    def run_online(self, coord):
+        self.env.cur_episode = 0
+        while not coord.should_stop():
+            time.sleep(30)
+            if self.global_counter.should_test():
+                rewards = []
+                global_step = self.global_counter.cur_step
+                for test_ind in range(self.test_num):
+                    cur_reward = self.perform(test_ind)
+                    self.env.terminate()
+                    rewards.append(cur_reward)
+                    log = {'agent': self.agent,
+                           'step': global_step,
+                           'test_id': test_ind,
+                           'reward': cur_reward}
+                    self.data.append(log)
+                avg_reward = np.mean(np.array(rewards))
+                self._add_summary(avg_reward, global_step)
+                logging.info('Testing: global step %d, avg R: %.2f' %
+                             (global_step, avg_reward))
+                # self.global_counter.update_test(avg_reward)
+        df = pd.DataFrame(self.data)
+        df.to_csv(self.output_path + 'train_reward.csv')
+
+
+class Predictor(Tester):
+    def __init__(self, env, model, demo=False, policy_type='default'):
+        self.env = env
+        self.model = model
+        self.agent = self.env.agent
+        self.env.train_mode = False
+        self.test_num = self.env.test_num
+        self.demo = demo
+        self.policy_type = policy_type
+
+    def run(self, state):
+        self.env.cur_episode = 0
+        time.sleep(1)
+        for test_ind in range(self.test_num):
+            action = self.inference(state, policy_type=self.policy_type)
+            time.sleep(2)
+
+        return action 
+
+
+class Evaluator(Tester):
+    def __init__(self, env, model, output_path, demo=False, policy_type='default'):
+        self.env = env
+        self.model = model
+        self.agent = self.env.agent
+        self.env.train_mode = False
+        self.test_num = self.env.test_num
+        self.output_path = output_path
+        self.demo = demo
+        self.policy_type = policy_type
+
+    def run(self):
+        is_record = True
+        record_stats = False
+        self.env.cur_episode = 0
+        self.env.init_data(is_record, record_stats, self.output_path)
+        time.sleep(1)
+        for test_ind in range(self.test_num):
+            reward, _ = self.perform(policy_type=self.policy_type)
+            logging.info('test %i, avg reward %.2f' % (test_ind, reward))
+            time.sleep(2)
+        self.env.output_data()