0
\$\begingroup\$

Related question: Extract the last row in a dataframe with a timestamp before some given time

This is part of the code for Deep Reinforcement Learning training on a standards-based gym env.

What I need:

env.train() executes faster

First I use _init_data to generate random demo data, then I use _init_datetime_selected to randomly take out the time to start training self.datetime_selected (for example, '2019-02-15 14:34:00').

I use _get_status to take the maximum time that is smaller than the selected time self.datetime_selected in different dfs, I take out the corresponding rows according to the maximum time, and generate these rows into df, which is the result I need.

The unit of duration is seconds, and the corresponding dict duration_to_period is converted to a period.

print(self.status)
duration             datetime        c01  ...        c08        c09        c10
0        60  2019-02-15 14:33:00  95.783565  ...  47.582137  28.811245  25.396333
1       300  2019-02-15 14:30:00  43.864518  ...  84.381690  88.221519  52.902785
2       900  2019-02-15 14:30:00  13.735799  ...  48.439307  97.794226  27.572249
3      1800  2019-02-15 14:30:00  75.068874  ...  61.772339  23.584858  49.992372
4      3600  2019-02-15 12:00:00  87.354660  ...  66.295853  31.067343   7.067540
5      7200  2019-02-15 12:00:00  83.048074  ...  76.678252  34.791799  51.644878
6     86400  2019-02-15 00:00:00  27.072804  ...  41.200409  12.177524  48.119748
7    172800  2019-02-15 00:00:00   6.386600  ...  46.880836   8.359451  81.630522
8    604800  2019-02-10 00:00:00  85.485440  ...  46.082749  24.165392  14.545734
9   2592000  2019-01-31 00:00:00  13.965196  ...  68.463874   9.728871  64.145777

Then use self.step to get the status of the next action until step_period is reached.

In my code, there is no action related code, just skip to the next minute:'2019-02-15 14:35:00'.

self.reset() starts the training of the new episode.

I wrote the relevant code in env.train() with comments: the neural network generates an action by predicting the status, self.step gets a new action according to the action and returns status, reward, done.

self._init_data is to generate demo data, others are production code.

I will use parallel env to speed up training, the relevant code is not included here.

In a production environment, MyEnv.episodes=4_000_000. On my computer, use %timeit env.train(1) to measure the speed, which is:

%timeit env.train(1)
1.56 s ± 24.3 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

Can my code speed be improved?

My code is as follows:

import pandas as pd
from tqdm import tqdm
import numpy as np
import datetime


class MyEnv:
    # The reason I put some properties in class properties is
    # that some properties need to be reassigned before the instance is initialized.
    durations: tuple = (60, 300, 900, 1800, 3600, 7200, 86400, 172800, 604800, 2592000)
    duration_to_freq: dict = {60: 'T', 120: '2T', 300: '5T', 900: '15T', 1800: '30T', 3600: 'H', 5400: '90T',
                              7200: '2H', 14400: '4H', 86400: 'B', 172800: '2B', 259200: '3B', 604800: 'W',
                              2592000: 'BM', 7776000: 'Q'}
    step_period: int = 120
    episodes: int = 100  # episodes=4_000_000 in my production application.

    rand = None

    data_start_dt: datetime.datetime = datetime.datetime(2018, 1, 20)
    data_end_dt: datetime.datetime = datetime.datetime.now()
    # Training_start_dt and data_start_dt are not the same because the time before a specific time needs to be selected.
    training_start_dt: datetime.datetime = datetime.datetime(2018, 2, 1)

    col: list = ['c01', 'c02', 'c03', 'c04', 'c05', 'c06', 'c07', 'c08', 'c09', 'c10', ]

    time_str: str = '%H:%M:%S'
    date_str: str = '%Y-%m-%d'
    datetime_str: str = '%Y-%m-%d %H:%M:%S'

    data_end_dt_str: str = None
    training_end_dt_str: str = None

    datetime_selected: str = None
    datetime_selected_idx: int = None
    training_datetime: np.ndarray = None
    len_training_datetime: int = None

    status: pd.DataFrame = None
    status_printed: bool = False

    def __init__(self):
        self.seed()
        self.training_end_dt: datetime.datetime = self.data_end_dt
        self.data_start_dt_str: str = datetime.datetime.strftime(self.data_start_dt, self.datetime_str)
        self.training_start_dt_str: str = datetime.datetime.strftime(self.training_start_dt, self.datetime_str)
        self.first_duration = self.durations[0]
        self.data_space: dict = {}
        self.training_data_space: dict = {}
        self.data_datetime: dict = {}

        self._init_data()  # This is the demo data.
        self.current_step: int = 0
        pass

    # Here is the demo data.
    def _init_data(self) -> None:
        for duration in self.durations:
            freq = self.duration_to_freq[duration]
            _df = pd.DataFrame(
                index=pd.date_range(self.data_start_dt, self.data_end_dt, freq=freq, name='datetime'))
            _df['duration'] = duration
            _df = _df.reset_index()

            if duration < 86400:
                _df['time'] = _df.datetime.dt.strftime(self.time_str)
                _df['weekday'] = _df.datetime.dt.strftime('%w')
                _df = _df[(_df.weekday > '0') & (_df.weekday < '6')]
                _df = _df[
                    ((_df.time > '10:00:00') & (_df.time <= '12:00:00'))
                    | ((_df.time > '14:00:00') & (_df.time <= '16:00:00'))
                    ]
                _df = _df.drop(columns=['time', 'weekday'])
                _df = _df.reset_index(drop=True)

            _df['datetime'] = _df.datetime.dt.strftime(self.datetime_str)
            _df[self.col] = (
                self.rand.uniform(low=1, high=100, size=len(_df) * len(self.col))
                    .reshape(len(_df), len(self.col))
            )

            self.data_space[duration] = _df

            datetime1 = _df.datetime.to_numpy()
            self.data_datetime[duration] = datetime1

            if duration == self.first_duration:
                self.training_datetime = datetime1[datetime1 >= self.training_start_dt_str]
                dt_str = datetime1[-1]
                dt = datetime.datetime.strptime(dt_str, self.datetime_str)
                self.data_end_dt = self.training_end_dt = dt
                self.data_end_dt_str = self.training_end_dt_str = dt_str
        return

    # The following code is all production code.
    def seed(self, seed: int = 0) -> None:
        self.rand = np.random.default_rng(seed=seed)
        np.random.seed(seed)

    def _init_datetime_selected(self) -> None:
        len_d = len(self.training_datetime)
        datetime_selected_idx = int(self.rand.integers(len_d))
        datetime_selected = self.training_datetime[datetime_selected_idx]
        self.len_training_datetime = len_d
        self.datetime_selected = datetime_selected
        self.datetime_selected_idx = datetime_selected_idx
        return

    def _datetime_step(self) -> None:
        len_d = self.len_training_datetime
        dt_idx = self.datetime_selected_idx + 1
        if dt_idx + 1 >= len_d:
            self.done = True
        datetime_selected = self.training_datetime[dt_idx]
        self.datetime_selected = datetime_selected
        self.datetime_selected_idx = dt_idx
        return

    def _get_status(self) -> pd.DataFrame:
        _df = self.status
        # self.datetime_selected will change with self.step().
        for idx in range(len(self.durations)):
            duration = self.durations[idx]
            dt1 = _df[_df.index == idx].datetime.to_numpy()[0]
            _df1 = self.data_space[duration]
            y = _df1.datetime.searchsorted(self.datetime_selected)
            while True:
                row = _df1.iloc[y, :]
                if row.datetime < self.datetime_selected:
                    break
                y -= 1
            if (dt1 is None) or (dt1 != row.datetime):
                _df = _df[_df.duration != duration]
                _df = _df.append(row)
            elif duration < 86400:
                break
        _df = _df.sort_values('duration').reset_index(drop=True)
        self.status = _df
        if not self.status_printed:
            print(_df)
            self.status_printed = True
        return self.status

    def step(self) -> tuple[list, bool]:
        self.current_step += 1
        self._datetime_step()
        _df = self._get_status()
        if self.current_step >= self.step_period or self.done:
            _done = True
        else:
            _done = False
        _status = list(_df.to_numpy().flatten())
        return _status, _done

    def _reset_properties(self) -> None:
        self.current_step = 0
        _df = pd.DataFrame([duration for duration in self.durations], columns=['duration'])
        _df['datetime'] = None
        _df[self.col] = None
        self.status = _df
        self.done = False

    def reset(self) -> list:
        self._reset_properties()
        self._init_datetime_selected()
        _df = self._get_status()
        return list(_df.to_numpy().flatten())

    def train(self, episodes: int = None) -> None:
        if episodes is None:
            episodes = self.episodes
        for _ in tqdm(range(episodes), total=episodes):
            status = env.reset()
            # Action is generated based on the state, these codes are omitted, and the production code is present.
            while True:
                # Go to the next step according to the action.
                # Reward etc. are omitted, production code has.

                # action=self.agent.predict(status)
                # next_status, reward, done, _ =env.step(action)
                # self.agent.learn(status,action,reward,next_status)
                # status=next_status

                status, done = env.step()
                if done:
                    # Agent learning is omitted, production code has.
                    break


if __name__ == '__main__':
    env = MyEnv()
    env.train()
\$\endgroup\$
0

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.