utils package

Submodules

utils.data module

class utils.data.SingleData(asset_code: str, info: Dict, hist: DataFrame, option_dates: Tuple, calls: Dict[str, DataFrame], puts: Dict[str, DataFrame])

Bases: object

the single data class

__init__(asset_code: str, info: Dict, hist: DataFrame, option_dates: Tuple, calls: Dict[str, DataFrame], puts: Dict[str, DataFrame]) → None

initialize the single data class

Parameters:

  • asset_code (str) – the asset code

  • info (Dict) – the asset info

  • hist (pd.DataFrame) – the asset history data

  • option_dates (Tuple) – the option dates for the asset

  • calls (Dict[str, pd.DataFrame]) – the calls data

  • puts (Dict[str, pd.DataFrame]) – the puts data

class utils.data.Data(data: Dict[str, SingleData] = {})

Bases: object

the Data manager for this project

__init__(data: Dict[str, SingleData] = {}) → None

initialize the Data manager

Parameters:

data (Dict[str, SingleData], optional) – a dict of SingleData. Defaults to {}.

check_time_list() → None

check the time list

delete_from_time(time: Timestamp) → None

delete data from the given time

Parameters:

time (pd.Timestamp) – the time to delete data from

uniform_time(time_zone: str = 'America/New_York') → None

uniform all data time to the given time zone

Parameters:

time_zone (str, optional) – the time zone. Defaults to “America/New_York”.

_get_time_list() → List[Timestamp]

get the list of all time index

Returns:

the list of all time index

Return type:

List[pd.Timestamp]

get_time_index(time: Timestamp) → int

get the index of the given time

Parameters:

time (pd.Timestamp) – the given timestamp

Returns:

the index of the given time

Return type:

int

time_dimension() → int

the number of time points

Returns:

the number of time points

Return type:

int

asset_dimension() → int

the number of assets

Returns:

the number of assets

Return type:

int

add_asset_data(asset_code: str, data: SingleData) → None

add asset data to the data manager

Parameters:

  • asset_code (str) – the new asset code

  • data (SingleData) – the asset data info

get_asset_data(asset_code: str) → SingleData

get the asset data

Parameters:

asset_code (str) – the asset code to get data for

Returns:

the asset data

Return type:

SingleData

get_asset_info(asset_code: str) → Dict[str, Any]

get the asset info

Parameters:

asset_code (str) – the asset code to get info for

Returns:

the asset info

Return type:

Dict[str, Any]

get_asset_hist(asset_code: str) → DataFrame

get the asset history data

Parameters:

asset_code (str) – the asset code to get history data for

Returns:

the asset history data

Return type:

pd.DataFrame

get_asset_hist_at_time(asset_code: str, time: Timestamp) → Optional[Series]

the asset history data at the given time

Parameters:

  • asset_code (str) – the asset code to get history data for

  • time (pd.Timestamp) – the timestamp to get history data for

Returns:

the asset history data at the given time

Return type:

Optional[pd.Series]

get_asset_option_dates(asset_code: str) → Tuple

get the asset option dates

Parameters:

asset_code (str) – the asset code to get option dates for

Returns:

the asset option dates

Return type:

Tuple

get_asset_calls(asset_code: str, date: str) → DataFrame

get the asset calls data

Parameters:

  • asset_code (str) – the asset code to get calls data for

  • date (str) – the date to get calls data for

Returns:

the asset calls data

Return type:

pd.DataFrame

get_asset_puts(asset_code: str, date: str) → DataFrame

get the asset puts data

Parameters:

  • asset_code (str) – the asset code to get puts data for

  • date (str) – the date to get puts data for

Returns:

the asset puts data

Return type:

pd.DataFrame

utils.data.load_data_object(base_data_path: str, asset_codes: List[str], start_date: Optional[str] = None, end_date: Optional[str] = None, interval: str = '1d', period: Optional[str] = None, reload: bool = False) → Data

load data for the given asset codes

Parameters:

  • base_data_path (str) – the base data path to save the asset data

  • asset_codes (List[str]) – the list of asset codes to use

  • start_date (Optional[str], optional) – start date. Defaults to None.

  • end_date (Optional[str], optional) – end date. Defaults to None.

  • interval (str, optional) – data interval, select from [“1m”, “2m”, “5m”, “15m”, “30m”, “60m”, “90m”, “1h”, “1d”, “5d”, “1wk”, “1mo”, “3mo”]. Defaults to “1d”.

  • period (Optional[str], optional) – fetch date period, select from [“1d”, “5d”, “1mo”, “3mo”, “6mo”, “1y”, “2y”, “5y”, “10y”, “ytd”, “max”]. Defaults to None.

  • reload (bool, optional) – whether or not to reload data. Defaults to False.

Returns:

the loaded data

Return type:

Dict[str, Dict[str, Any]]

utils.data.get_asset_data(base_data_path: str, asset_code: str, start_date: Optional[str] = None, end_date: Optional[str] = None, interval: str = '1d', period: Optional[str] = None, reload: bool = False) → Tuple[Dict, DataFrame, Tuple, Dict[str, DataFrame], Dict[str, DataFrame]]

get asset data from Yahoo Finance

Parameters:

  • base_data_path (str) – base data path to save the asset data

  • asset_code (str) – asset code to fetch data for. E.g. AAPL, TSLA, MSFT, etc.

  • start_date (Optional[str], optional) – start date. Defaults to None.

  • end_date (Optional[str], optional) – end date. Defaults to None.

  • interval (str, optional) – data interval, select from [“1m”, “2m”, “5m”, “15m”, “30m”, “60m”, “90m”, “1h”, “1d”, “5d”, “1wk”, “1mo”, “3mo”]. Defaults to “1d”.

  • period (Optional[str], optional) – fetch date period, select from [“1d”, “5d”, “1mo”, “3mo”, “6mo”, “1y”, “2y”, “5y”, “10y”, “ytd”, “max”]. Defaults to None.

  • reload (bool, optional) – whether or not to reload data. Defaults to False.

Raises:

  • ValueError – start_date and end_date must be used together

  • ValueError – start_date and end_date cannot be used with period

  • ValueError – at least one of start_date, end_date, or period must be used

  • ValueError – no data found for asset_code

Returns:

asset info, asset history, options dates, calls, puts

Return type:

Tuple[Dict, pd.DataFrame, tuple, Dict, Dict]

utils.data.get_and_save_asset_data(base_data_path: str, asset_code: str, start_date: Optional[str] = None, end_date: Optional[str] = None, interval: str = '1d', period: Optional[str] = None, reload: bool = False) → Tuple[Dict, DataFrame, Tuple, Dict[str, DataFrame], Dict[str, DataFrame], str]

get and save asset data

Parameters:

  • base_data_path (str) – base data path to save the asset data

  • asset_code (str) – asset code to fetch data for. E.g. AAPL, TSLA, MSFT, etc.

  • start_date (Optional[str], optional) – start date . Defaults to None.

  • end_date (Optional[str], optional) – end date. Defaults to None.

  • interval (str, optional) – data interval, select from [“1m”, “2m”, “5m”, “15m”, “30m”, “60m”, “90m”, “1h”, “1d”, “5d”, “1wk”, “1mo”, “3mo”]. Defaults to “1d”.

  • period (Optional[str], optional) – fetch date period, select from [“1d”, “5d”, “1mo”, “3mo”, “6mo”, “1y”, “2y”, “5y”, “10y”, “ytd”, “max”]. Defaults to None.

  • reload (bool, optional) – whether or not to reload data. Defaults to False.

Returns:

asset info, asset history, options dates, calls, puts, base path

Return type:

Tuple[Dict, pd.DataFrame, Tuple, Dict, Dict, str]

utils.data.save_asset_data(base_data_path: str, asset_code: str, start_date: Optional[str] = None, end_date: Optional[str] = None, interval: str = '1d', period: Optional[str] = None, reload: bool = False) → str

save asset data

Parameters:

  • base_data_path (str) – base data path to save the asset data

  • asset_code (str) – asset code to fetch data for. E.g. AAPL, TSLA, MSFT, etc.

  • start_date (Optional[str], optional) – start date . Defaults to None.

  • end_date (Optional[str], optional) – end date. Defaults to None.

  • interval (str, optional) – data interval, select from [“1m”, “2m”, “5m”, “15m”, “30m”, “60m”, “90m”, “1h”, “1d”, “5d”, “1wk”, “1mo”, “3mo”]. Defaults to “1d”.

  • period (Optional[str], optional) – fetch date period, select from [“1d”, “5d”, “1mo”, “3mo”, “6mo”, “1y”, “2y”, “5y”, “10y”, “ytd”, “max”]. Defaults to None.

  • reload (bool, optional) – whether or not to reload data. Defaults to False.

Returns:

base path

Return type:

str

utils.data.load_single_data_from_local(base_data_path: str, asset_code: str, start_date: Optional[str] = None, end_date: Optional[str] = None, interval: str = '1d', period: Optional[str] = None) → Tuple[Dict, DataFrame, Tuple, Dict[str, DataFrame], Dict[str, DataFrame]]

load data from local file

Parameters:

  • base_data_path (str) – base data path to save the asset data

  • asset_code (str) – asset code to fetch data for. E.g. AAPL, TSLA, MSFT, etc.

  • start_date (Optional[str], optional) – start date . Defaults to None.

  • end_date (Optional[str], optional) – end date. Defaults to None.

  • interval (str, optional) – data interval, select from [“1m”, “2m”, “5m”, “15m”, “30m”, “60m”, “90m”, “1h”, “1d”, “5d”, “1wk”, “1mo”, “3mo”]. Defaults to “1d”.

  • period (Optional[str], optional) – fetch date period, select from [“1d”, “5d”, “1mo”, “3mo”, “6mo”, “1y”, “2y”, “5y”, “10y”, “ytd”, “max”]. Defaults to None.

Returns:

asset info, asset history, options dates, calls, puts

Return type:

Tuple[ Dict, pd.DataFrame, Tuple, Dict[str, pd.DataFrame], Dict[str, pd.DataFrame]]

utils.data._save_asset_info(base_path: str, info: Dict) → None

save asset info to local file

Parameters:

  • base_path (str) – the base path to save the asset data

  • info (Dict) – the asset info to save

utils.data._get_asset_info(base_path: str) → Dict

get asset info from local file

Parameters:

base_path (str) – the base path to save the asset data

Returns:

the asset info

Return type:

Dict

utils.data._save_asset_hist(base_path: str, hist: DataFrame) → None

save asset history to local file

Parameters:

  • base_path (str) – the base path to save the asset data

  • hist (pd.DataFrame) – the asset history to save

utils.data._get_asset_hist(base_path: str) → DataFrame

get asset history from local file

Parameters:

base_path (str) – the base path to save the asset data

Returns:

the asset history

Return type:

pd.DataFrame

utils.data._save_asset_options_dates(base_path: str, option_dates: Tuple) → None

save asset option dates to local file

Parameters:

  • base_path (str) – the base path to save the asset data

  • option_dates (Tuple) – the asset option dates to save

utils.data._get_asset_options_dates(base_path: str) → Tuple

get asset option dates from local file

Parameters:

base_path (str) – the base path to save the asset data

Returns:

the asset option dates

Return type:

Tuple

utils.data._save_asset_calls(base_path: str, calls: Dict[str, DataFrame]) → None

save asset calls to local file

Parameters:

  • base_path (str) – the base path to save the asset data

  • calls (Dict[str, pd.DataFrame]) – the asset calls to save

utils.data._get_asset_calls(base_path: str, dates: Tuple) → Dict[str, DataFrame]

get asset calls from local file

Parameters:

  • base_path (str) – the base path to save the asset data

  • dates (Tuple) – the dates to get calls data for

Returns:

the asset calls

Return type:

Dict[str, pd.DataFrame]

utils.data._save_asset_puts(base_path: str, puts: Dict[str, DataFrame]) → None

save asset puts to local file

Parameters:

  • base_path (str) – the base path to save the asset data

  • puts (Dict[str, pd.DataFrame]) – the asset puts to save

utils.data._get_asset_puts(base_path: str, dates: Tuple) → Dict[str, DataFrame]

get asset puts from local file

Parameters:

  • base_path (str) – the base path to save the asset data

  • dates (Tuple) – the dates to get puts data for

Returns:

the asset puts

Return type:

Dict[str, pd.DataFrame]

utils.data._data_path(base_data_path: str, asset_code: str, start_date: Optional[str] = None, end_date: Optional[str] = None, interval: str = '1d', period: Optional[str] = None) → str

get data base path to save the asset data

Parameters:

  • base_data_path (str) – base data path to save the asset data

  • asset_code (str) – asset code to fetch data for. E.g. AAPL, TSLA, MSFT, etc.

  • start_date (Optional[str], optional) – start date . Defaults to None.

  • end_date (Optional[str], optional) – end date. Defaults to None.

  • interval (str, optional) – data interval, select from [“1m”, “2m”, “5m”, “15m”, “30m”, “60m”, “90m”, “1h”, “1d”, “5d”, “1wk”, “1mo”, “3mo”]. Defaults to “1d”.

  • period (Optional[str], optional) – fetch date period, select from [“1d”, “5d”, “1mo”, “3mo”, “6mo”, “1y”, “2y”, “5y”, “10y”, “ytd”, “max”]. Defaults to None.

Returns:

data base path to save the asset data

Return type:

str

utils.file module

utils.file.create_path_recursively(path: str) → None

Recursively create a path if it doesn’t exist.

Parameters:

path (str) – the path to create

utils.logging module

class utils.logging.ColoredFormatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)

Bases: Formatter

A logging formatter that colors the output

COLORS = {'CRITICAL': '\x1b[95m', 'DEBUG': '\x1b[94m', 'ENDC': '\x1b[0m', 'ERROR': '\x1b[91m', 'INFO': '\x1b[92m', 'WARNING': '\x1b[93m'}

the colors for different log levels

format(record: LogRecord) → str

format the log record

Parameters:

record (logging.LogRecord) – the log record

Returns:

the formatted log record

Return type:

str

utils.logging.set_up_logging(level: str = 'INFO') → None

set up basic logging configuration

Parameters:

level (str, optional) – the verbose level. Defaults to “INFO”.

utils.logging.get_logger(name: str) → Logger

get a logger

Parameters:

name (str) – the name of the logger

Returns:

the logger

Return type:

logging.Logger

utils.replay module

class utils.replay.Replay(args: Namespace)

Bases: object

The replay memory

static add_args(parser: ArgumentParser) → None

add arguments to the parser

to add arguments to the parser, modify the method as follows:

@staticmethod
def add_args(parser: argparse.ArgumentParser) -> None:
    parser.add_argument(
        ...
    )

then add arguments to the parser

Parameters:

parser (argparse.ArgumentParser) – the parser to add arguments to

__init__(args: Namespace) → None

initialize the replay memory

Parameters:

args (argparse.Namespace) – the arguments

remember(experience: Any) → None

remember an experience

Parameters:

experience (Any) – the experience to remember

sample(size: Optional[int] = None, interval: Optional[int] = None) → Any

randomly sample a batch of experiences

Parameters:

  • size (Optional[int], optional) – the size of the batch. Defaults to None.

  • interval (Optional[int], optional) – the interval to ignore the last few experiences. Defaults to None.

Returns:

the batch of experiences

Return type:

Any

reset() → None

reset the replay memory to an empty state

has_enough_samples(interval: Optional[int] = None) → bool

return if the replay memory has enough samples

Parameters:

interval (Optional[int], optional) – the interval to ignore the last few experiences. Defaults to None.

Returns:

whether the replay memory has enough samples

Return type:

bool