pyqstrat package

Submodules

pyqstrat.pq_utils module

exception pyqstrat.pq_utils.PQException

Bases: Exception

class pyqstrat.pq_utils.Paths(base_path=None)

Bases: object

Conventions for where to read / write data and reports

create()

Return type:

None

pyqstrat.pq_utils.assert_(condition, msg=None)

Like a python assert but raises an exception that is not turned off by using the python optimization switch

Return type:

None

pyqstrat.pq_utils.bootstrap_ci(a, ci_level=0.95, n=1000, func=<function mean>)

Non parametric bootstrap for confidence intervals :type a: ndarray :param a: The data to bootstrap from :type ci_level: float :param ci_level: The confidence interval level, e.g. 0.95 for 95%. Default 0.95 :type n: int :param n: Number of boostrap iterations. Default 1000 :type func: Callable[[ndarray], ndarray] :param func: The function to use, e.g np.mean or np.median. Default np.mean

Return type:

tuple[float, float]

Returns:

A tuple containing the lower and upper ci

>>> np.random.seed(0)
>>> x = np.random.uniform(high=10, size=100000)
>>> assert np.allclose(bootstrap_ci(x), (4.9773159, 5.010328))

pyqstrat.pq_utils.day_of_week_num(a)

From https://stackoverflow.com/questions/52398383/finding-day-of-the-week-for-a-datetime64 Get day of week for a numpy array of datetimes Monday is 0, Sunday is 6

Parameters:

a (datetime64 | ndarray) – numpy datetime64 or array of datetime64

Returns:

Monday is 0, Sunday is 6

Return type:

int or numpy ndarray of int

>>> day_of_week_num(np.datetime64('2015-01-04'))
6

pyqstrat.pq_utils.day_symbol(day_int)

Return type:

str | ndarray

pyqstrat.pq_utils.find_in_subdir(dir, filename)

Find relative path of a file in a subdirectory

Return type:

str

pyqstrat.pq_utils.get_child_logger(child_name)

Return type:

Logger

pyqstrat.pq_utils.get_config()

Load config data from yaml file and returns it as a dict. This first loads values from a config file called pyqstrat.yml in your home directory. Next it looks for a file called pyqstrat.yml in your local working directory. If found, it overrides any values in the config data from any data it finds in this file.

Return type:

dict[str, Any]

pyqstrat.pq_utils.get_empty_np_value(np_dtype)

Get empty value for a given numpy datatype >>> a = np.array([‘2018-01-01’, ‘2018-01-03’], dtype = ‘M8[D]’) >>> get_empty_np_value(a.dtype) numpy.datetime64(‘NaT’)

Return type:

Any

pyqstrat.pq_utils.get_main_logger()

Return type:

Logger

pyqstrat.pq_utils.get_paths(base_path=None)

Return type:

Paths

pyqstrat.pq_utils.get_temp_dir()

Return type:

str

pyqstrat.pq_utils.has_display()

Useful for running on a headless machine such as a remote server so we don’t try to show graphs etc during unit tests

Return type:

bool

pyqstrat.pq_utils.in_debug()

Return type:

bool

pyqstrat.pq_utils.in_ipython()

Whether we are running in an ipython (or Jupyter) environment

Return type:

bool

pyqstrat.pq_utils.infer_compression(input_filename)

Infers compression for a file from its suffix. For example, given “/tmp/hello.gz”, this will return “gzip” >>> infer_compression(“/tmp/hello.gz”) ‘gzip’ >>> infer_compression(“/tmp/abc.txt”) is None True

Return type:

Optional[str]

pyqstrat.pq_utils.infer_frequency(timestamps)

Returns most common frequency of date differences as a fraction of days :type timestamps: ndarray :param timestamps: A numpy array of monotonically increasing datetime64

>>> timestamps = np.array(['2018-01-01 11:00:00', '2018-01-01 11:15', '2018-01-01 11:30', '2018-01-01 11:35'], dtype = 'M8[ns]')
>>> infer_frequency(timestamps)  
Traceback (most recent call last):
 ...
PQException: could not infer frequency from timestamps...
>>> timestamps = np.array(['2018-01-01 11:00', '2018-01-01 11:15', '2018-01-01 11:30', '2018-01-01 11:35', '2018-01-01 11:50'], dtype = 'M8[ns]')
>>> print(round(infer_frequency(timestamps), 8))
0.01041667
:rtype: :py:class:`float`

>>> timestamps = np.array(['2015-01-01', '2015-03-01', '2015-05-01', '2015-07-01', '2015-09-01'], dtype='M8[D]')
>>> assert math.isclose(infer_frequency(timestamps), 60)

pyqstrat.pq_utils.is_newer(filename, ref_filename)

whether filename ctime (modfication time) is newer than ref_filename or either file does not exist >>> import time >>> import tempfile >>> temp_dir = tempfile.gettempdir() >>> touch(f’{temp_dir}/x.txt’) >>> time.sleep(0.1) >>> touch(f’{temp_dir}/y.txt’) >>> is_newer(f’{temp_dir}/y.txt’, f’{temp_dir}/x.txt’) True >>> touch(f’{temp_dir}/y.txt’) >>> time.sleep(0.1) >>> touch(f’{temp_dir}/x.txt’) >>> is_newer(f’{temp_dir}/y.txt’, f’{temp_dir}/x.txt’) False

Return type:

bool

pyqstrat.pq_utils.linear_interpolate(a1, a2, x1, x2, x)

>>> assert(linear_interpolate(3, 4, 8, 10, 8.9) == 3.45)
>>> assert(linear_interpolate(3, 3, 8, 10, 8.9) == 3)
>>> assert(np.isnan(linear_interpolate(3, 4, 8, 8, 8.9)))
>>> x = linear_interpolate(
...    np.array([3., 3.]),
...    np.array([4., 3.]),
...    np.array([8., 8.]),
...    np.array([10, 8.]),
:rtype: :py:class:`~numpy.ndarray` | :py:class:`float`

… np.array([8.9, 8.])) >>> assert(np.allclose(x, np.array([3.45, 3.])))

pyqstrat.pq_utils.millis_since_epoch(dt)

Given a python datetime, return number of milliseconds between the unix epoch and the datetime. Returns a float since it can contain fractions of milliseconds as well >>> millis_since_epoch(datetime.datetime(2018, 1, 1)) 1514764800000.0

Return type:

float

pyqstrat.pq_utils.monotonically_increasing(array)

Returns True if the array is monotonically_increasing, False otherwise

>>> monotonically_increasing(np.array(['2018-01-02', '2018-01-03'], dtype = 'M8[D]'))
True
:rtype: :py:class:`bool`

>>> monotonically_increasing(np.array(['2018-01-02', '2018-01-02'], dtype = 'M8[D]'))
False

pyqstrat.pq_utils.nan_to_zero(array)

Converts any nans in a numpy float array to 0

Return type:

ndarray

pyqstrat.pq_utils.np_bucket(a, buckets, default_value=0, side='mid')

Given a numpy array and a sorted list of buckets, assign each element to a bucket.

Parameters:

• a (np.ndarray) – The numpy array of values

• buckets (list[Any]) – (list) List of buckets

• default_value – Used when we cannot assign an element to any bucket if side is ‘left’ or ‘right’

• side (str) – If set to mid, we use the midpoint between buckets to assign elements ‘left’, assignment <= element ‘right’, assignment >= element Default: ‘mid’

Return type:

ndarray

Returns:

np.ndarray of same length as a

>>> a = np.array([1, 5, 18, 3, 6, 10, 4])
>>> buckets = [4, 8, 12]
>>> assert np.allclose(np_bucket(a, buckets, side='left'), np.array([0,  4, 12,  0,  4,  8,  4]))
>>> assert np.allclose(np_bucket(a, buckets, default_value=25, side='right'), np.array([4,  8, 25,  4,  8, 12,  4]))
>>> assert np.allclose(np_bucket(a, buckets), np.array([4,  4, 12,  4,  8, 12,  4]))

pyqstrat.pq_utils.np_find_closest(a, v)

From https://stackoverflow.com/questions/8914491/finding-the-nearest-value-and-return-the-index-of-array-in-python Find index of closest value to array v in array a. Returns an array of the same size as v a must be sorted >>> assert(all(np_find_closest(np.array([3, 4, 6]), np.array([4, 2])) == np.array([1, 0])))

Return type:

int | ndarray

pyqstrat.pq_utils.np_inc_dates(dates, num_days=1)

Increment the given date array so each cell gets the next higher value in this array >>> dates = np.array([‘2021-06-01’, ‘2021-06-01’, ‘2021-08-01’, ‘2021-04-01’], dtype=’M8[D]’) >>> check = np.array([dates[2], dates[2], np.datetime64(‘nat’), dates[0]]) >>> assert np.array_equal(np_inc_dates(dates, 1), … np.array([‘2021-08-01’, ‘2021-08-01’, ‘NaT’, ‘2021-06-01’], dtype=’M8[D]’), equal_nan=True) >>> assert np.array_equal(np_inc_dates(dates, 2), … np.array([‘NaT’, ‘NaT’, ‘NaT’, ‘2021-08-01’], dtype=’M8[D]’), equal_nan=True) >>> assert np.array_equal(np_inc_dates(dates, -1), … np.array([‘2021-04-01’, ‘2021-04-01’, ‘2021-06-01’, ‘NaT’], dtype=’M8[D]’), equal_nan=True) >>> assert np.array_equal(np_inc_dates(dates, -2), … np.array([‘NaT’, ‘NaT’, ‘2021-04-01’, ‘NaT’], dtype=’M8[D]’), equal_nan=True)

Return type:

ndarray

pyqstrat.pq_utils.np_indexof(array, value)

Get index of a value in a numpy array. Returns -1 if the value does not exist.

Return type:

int

pyqstrat.pq_utils.np_indexof_sorted(array, value)

Get index of a value in a sorted numpy array. Returns -1 if the value does not exist a = np.array([1, 2, 3, 4]) assert(np_indexof_sorted(a, 3) == 2) assert(np.indexof_sorted(a, 8) == -1) assert(np.indexof_sorted(a, 0) == -1)

Return type:

int

pyqstrat.pq_utils.np_parse_array(s, dtype=<class 'float'>)

Create a 1 or 2 d numpy array from a string that looks like: [[2. 5. 3. 0. 0.] [3. 5. 0. 4. 3.]] or [2. 5. 3. 0. 8.]

>>> x = np_parse_array('[[2. 5. 3. 0. 0.]\n [3. 5. 0. 4. 3.]]')
>>> assert np.allclose(x, np.array([[2., 5., 3., 0., 0.], [3., 5., 0., 4., 3.]]))
:rtype: :py:class:`~numpy.ndarray`

>>> x = np_parse_array('[3 4. 5]')
>>> assert np.allclose(x, np.array([3, 4., 5]))

pyqstrat.pq_utils.np_rolling_window(a, window)

For applying rolling window functions to a numpy array See: https://stackoverflow.com/questions/6811183/rolling-window-for-1d-arrays-in-numpy >>> print(np.std(np_rolling_window(np.array([1, 2, 3, 4]), 2), 1)) [0.5 0.5 0.5]

Return type:

ndarray

pyqstrat.pq_utils.np_round(a, clip)

Round all elements in an array to the nearest clip

Parameters:

• a (ndarray) – array with elements to round

• clip (float) – rounding value

>>> np_round(15.8, 0.25)
15.75

pyqstrat.pq_utils.np_uniques(arrays)

Given a list of numpy arrays that may have different datatype, generate a structured numpy array with sorted, unique elements from that list >>> array1 = np.array([‘2018-01-02’, ‘2018-01-03’, ‘2018-01-02’, ‘2018-01-03’], dtype=’M8[D]’) >>> array2 = np.array([‘P’, ‘P’, ‘P’, ‘C’]) >>> x = np_uniques([array1, array2]) >>> assert len(x) == 3 >>> assert x[0][0] == np.datetime64(‘2018-01-02’) >>> assert x[0][1] == ‘P’

Return type:

ndarray

pyqstrat.pq_utils.percentile_of_score(a)

For each element in a, find the percentile of a its in. From stackoverflow.com/a/29989971/5351549 Like scipy.stats.percentileofscore but runs in O(n log(n)) time. >>> a = np.array([4, 3, 1, 2, 4.1]) >>> percentiles = percentile_of_score(a) >>> assert(all(np.isclose(np.array([ 75., 50., 0., 25., 100.]), percentiles)))

Return type:

Optional[ndarray]

pyqstrat.pq_utils.remove_dups(lst, key_func=None)

Remove duplicates from a list :type lst: list[Any] :param lst: list to remove duplicates from :type key_func: Optional[Callable[[Any], Any]] :param key_func: A function that takes a list element and converts it to a key for detecting dups

Return type:

list[Any]

Returns:

A list with duplicates removed. This is stable in the sense that original list elements will retain their order

>>> print(remove_dups(['a', 'd', 'a', 'c']))
['a', 'd', 'c']
>>> print(remove_dups(['a', 'd', 'A']))
['a', 'd', 'A']
>>> print(remove_dups(['a', 'd', 'A'], key_func = lambda e: e.upper()))
['a', 'd']

pyqstrat.pq_utils.resample_trade_bars(df, sampling_frequency, resample_funcs=None)

Downsample trade bars using sampling frequency

Parameters:

• df (pd.DataFrame) – Must contain an index of numpy datetime64 type which is monotonically increasing sampling_frequency (str): See pandas frequency strings

• str (resample_funcs (dict of) – int): a dictionary of column name -> resampling function for any columns that are custom defined. Default None. If there is no entry for a custom column, defaults to ‘last’ for that column

Returns:

Resampled dataframe

Return type:

pd.DataFrame

>>> import math
>>> df = pd.DataFrame({'date': np.array(['2018-01-08 15:00:00', '2018-01-09 13:30:00', '2018-01-09 15:00:00', '2018-01-11 15:00:00'], dtype = 'M8[ns]'),
...          'o': np.array([8.9, 9.1, 9.3, 8.6]),
...          'h': np.array([9.0, 9.3, 9.4, 8.7]),
...          'l': np.array([8.8, 9.0, 9.2, 8.4]),
...          'c': np.array([8.95, 9.2, 9.35, 8.5]),
...          'v': np.array([200, 100, 150, 300]),
...          'x': np.array([300, 200, 100, 400])
...         })
>>> df['vwap'] =  0.5 * (df.l + df.h)
>>> df.set_index('date', inplace = True)
>>> df = resample_trade_bars(df, sampling_frequency = 'D', resample_funcs={'x': lambda df,
...   sampling_frequency: df.x.resample(sampling_frequency).agg('mean')})
>>> assert(len(df) == 4)
>>> assert(math.isclose(df.vwap.iloc[1], 9.24))
>>> assert(np.isnan(df.vwap.iloc[2]))
>>> assert(math.isclose(df.l[3], 8.4))

pyqstrat.pq_utils.resample_ts(dates, values, sampling_frequency)

Downsample a tuple of datetimes and value arrays using sampling frequency, using the last value if it does not exist at the bin edge. See pandas.Series.resample

Parameters:

• dates (ndarray) – a numpy datetime64 array

• values (ndarray) – a numpy array

• sampling_frequency (str) – See pandas frequency strings

Return type:

tuple[ndarray, ndarray]

Returns:

Resampled tuple of datetime and value arrays

pyqstrat.pq_utils.resample_vwap(df, sampling_frequency)

Compute weighted average of vwap given higher frequency vwap and volume

Return type:

Optional[ndarray]

pyqstrat.pq_utils.series_to_array(series)

Convert a pandas series to a numpy array. If the object is not a pandas Series return it back unchanged

Return type:

ndarray

pyqstrat.pq_utils.set_defaults(df_float_sf=9, df_display_max_rows=200, df_display_max_columns=99, np_seterr='raise')

Set some display defaults to make it easier to view dataframes and graphs.

Parameters:

• df_float_sf (int) – Number of significant figures to show in dataframes (default 4). Set to None to use pandas defaults

• df_display_max_rows (int) – Number of rows to display for pandas dataframes when you print them (default 200). Set to None to use pandas defaults

• df_display_max_columns (int) – Number of columns to display for pandas dataframes when you print them (default 99). Set to None to use pandas defaults

• np_seterr (str) – Error mode for numpy warnings. See numpy seterr function for details. Set to None to use numpy defaults

• jupyter_multiple_display – If set, and you have multiple outputs in a Jupyter cell, output will contain all of them. Default True

Return type:

None

pyqstrat.pq_utils.set_ipython_defaults(jupyter_multiple_display=True)

Return type:

None

pyqstrat.pq_utils.shift_np(array, n, fill_value=None)

Similar to pandas.Series.shift but works on numpy arrays.

Parameters:

• array (ndarray) – The numpy array to shift

• n (int) – Number of places to shift, can be positive or negative

• fill_value (Optional[Any]) – After shifting, there will be empty slots left in the array. If set, fill these with fill_value. If fill_value is set to None (default), we will fill these with False for boolean arrays, np.nan for floats

Return type:

ndarray

pyqstrat.pq_utils.str2date(s)

Converts a string like “2008-01-15 15:00:00” to a numpy datetime64. If s is not a string, return s back

Return type:

datetime64

pyqstrat.pq_utils.strtup2date(tup)

Converts a string tuple like (“2008-01-15”, “2009-01-16”) to a numpy datetime64 tuple. If the tuple does not contain strings, return it back unchanged

Return type:

tuple[Optional[datetime64], Optional[datetime64]]

pyqstrat.pq_utils.to_csv(df, file_name, index=False, compress=False, *args, **kwargs)

Creates a temporary file then renames to the permanent file so we don’t have half written files. Also optionally compresses using the xz algorithm

Return type:

None

pyqstrat.pq_utils.touch(fname, mode=438, dir_fd=None, **kwargs)

replicate unix touch command, i.e create file if it doesn’t exist, otherwise update timestamp

Return type:

None

pyqstrat.pq_utils.try_frequency(timestamps, period, threshold)

Return type:

float

pyqstrat.pq_utils.zero_to_nan(array)

Converts any zeros in a numpy array to nans

Return type:

ndarray

pyqstrat.pq_types module

class pyqstrat.pq_types.Contract(symbol, contract_group, expiry, multiplier, components, properties)

Bases: object

static clear_cache()

Remove our cache of contract groups

Return type:

None

components: list[tuple[Contract, float]]

contract_group: ContractGroup

static create(symbol, contract_group=None, expiry=None, multiplier=1.0, components=None, properties=None)

Parameters:

• symbol (str) – A unique string reprenting this contract. e.g IBM or ESH9

• contract_group (Optional[ContractGroup]) – We sometimes need to group contracts for calculating PNL, for example, you may have a strategy which has options and a future or equity used to hedge delta. In this case, you will be trading different symbols over time as options and futures expire, but you may want to track PNL for each leg using a contract group for each leg. So you could create contract groups ‘OPTIONS’ and one for ‘HEDGES’

• expiry (Optional[datetime64]) – In the case of a future or option, the date and time when the contract expires. For equities and other non expiring contracts, set this to None. Default None.

• multiplier (float) – If the market price convention is per unit, and the unit is not the same as contract size, set the multiplier here. For example, for E-mini contracts, each contract is 50 units and the price is per unit, so multiplier would be 50. Default 1

• properties (Optional[SimpleNamespace]) – Any data you want to store with this contract. For example, you may want to store option strike. Default None

Return type:

Contract

static exists(name)

Return type:

bool

expiry: Optional[datetime64]

static get(name)

Returns an existing contrat or none if it does not exist

Return type:

Optional[Contract]

static get_or_create(symbol, contract_group=None, expiry=None, multiplier=1.0, components=None, properties=None)

Return type:

Contract

is_basket()

Return type:

bool

multiplier: float

properties: SimpleNamespace

A contract such as a stock, option or a future that can be traded

symbol: str

class pyqstrat.pq_types.ContractGroup(name)

Bases: object

A way to group contracts for figuring out which indicators, rules and signals to apply to a contract and for PNL reporting

add_contract(contract)

clear()

Remove all contracts

Return type:

None

static clear_cache()

Return type:

None

contracts: dict[str, Contract]

static exists(name)

Return type:

bool

static get(name)

Return type:

ContractGroup

Create a contract group if it does not exist, or returns an existing one

Args:

name: Name of the group

get_contract(symbol)

Return type:

Optional[Contract]

get_contracts()

Return type:

list[Contract]

static get_default()

Return type:

ContractGroup

name: str

class pyqstrat.pq_types.LimitOrder(*, contract, timestamp=numpy.datetime64('NaT'), qty=nan, reason_code='', time_in_force=TimeInForce.FOK, properties=<factory>, status=OrderStatus.OPEN, limit_price)

Bases: Order

limit_price: float

class pyqstrat.pq_types.MarketOrder(*, contract, timestamp=numpy.datetime64('NaT'), qty=nan, reason_code='', time_in_force=TimeInForce.FOK, properties=<factory>, status=OrderStatus.OPEN)

Bases: Order

class pyqstrat.pq_types.Order(*, contract, timestamp=numpy.datetime64('NaT'), qty=nan, reason_code='', time_in_force=TimeInForce.FOK, properties=<factory>, status=OrderStatus.OPEN)

Bases: object

Parameters:

• contract (Contract) – The contract this order is for

• timestamp (datetime64) – Time the order was placed

• qty (float) – Number of contracts or shares. Use a negative quantity for sell orders

• reason_code (str) – The reason this order was created. Default ‘’

• properties (SimpleNamespace) – Any order specific data we want to store. Default None

• status (OrderStatus) – Status of the order, “open”, “filled”, etc. Default “open”

cancel()

Return type:

None

contract: Contract

fill(fill_qty=nan)

Return type:

None

is_open()

Return type:

bool

properties: SimpleNamespace

qty: float = nan

reason_code: str = ''

request_cancel()

Return type:

None

status: OrderStatus = 1

time_in_force: TimeInForce = 1

timestamp: datetime64 = numpy.datetime64('NaT')

class pyqstrat.pq_types.OrderStatus(value)

Bases: Enum

Enum for order status

CANCELLED = 5

CANCEL_REQUESTED = 4

FILLED = 3

OPEN = 1

PARTIALLY_FILLED = 2

class pyqstrat.pq_types.Price(timestamp, bid, ask, bid_size, ask_size, valid=True, properties=None)

Bases: object

>>> price = Price(datetime.datetime(2020, 1, 1), 15.25, 15.75, 189, 300)
>>> print(price)
15.25@189/15.75@300
>>> price.properties = SimpleNamespace(delta = -0.3)
>>> price.valid = False
>>> print(price)
15.25@189/15.75@300 delta: -0.3 invalid
>>> print(price.mid())
15.5

ask: float

ask_size: int

bid: float

bid_size: int

static invalid()

Return type:

Price

mid()

Return type:

float

properties: Optional[SimpleNamespace] = None

set_property(name, value)

Return type:

None

spread()

Return type:

float

timestamp: datetime

valid: bool = True

vw_mid()

Volume weighted mid >>> price = Price(datetime.datetime(2020, 1, 1), 15.25, 15.75, 189, 300) >>> print(f’{price.vw_mid():.4f}’) 15.4433 >>> price.bid_size = 0 >>> price.ask_size = 0 >>> assert math.isnan(price.vw_mid())

Return type:

float

class pyqstrat.pq_types.RollOrder(*, contract, timestamp=numpy.datetime64('NaT'), qty=nan, reason_code='', time_in_force=TimeInForce.FOK, properties=<factory>, status=OrderStatus.OPEN, close_qty, reopen_qty)

Bases: Order

close_qty: float

reopen_qty: float

class pyqstrat.pq_types.RoundTripTrade(contract, entry_order, exit_order, entry_timestamp, exit_timestamp, qty, entry_price, exit_price, entry_reason, exit_reason, entry_commission, exit_commission, entry_properties=<factory>, exit_properties=<factory>, net_pnl=nan)

Bases: object

contract: Contract

entry_commission: float

entry_order: Order

entry_price: float

entry_properties: SimpleNamespace

entry_reason: Optional[str]

entry_timestamp: datetime64

exit_commission: float

exit_order: Optional[Order]

exit_price: float

exit_properties: SimpleNamespace

exit_reason: Optional[str]

exit_timestamp: datetime64

net_pnl: float = nan

qty: int

class pyqstrat.pq_types.StopLimitOrder(*, contract, timestamp=numpy.datetime64('NaT'), qty=nan, reason_code='', time_in_force=TimeInForce.FOK, properties=<factory>, status=OrderStatus.OPEN, trigger_price, limit_price=nan, triggered=False)

Bases: Order

Used for stop loss or stop limit orders. The order is triggered when price goes above or below trigger price, depending on whether this is a short

or long order. Becomes either a market or limit order at that point, depending on whether you set the limit price or not.

Parameters:

• trigger_price (float) – Order becomes a market or limit order if price crosses trigger_price.

• limit_price (float) – If not set (default), order becomes a market order when price crosses trigger price. Otherwise it becomes a limit order. Default np.nan

limit_price: float = nan

trigger_price: float

triggered: bool = False

class pyqstrat.pq_types.TimeInForce(value)

Bases: Enum

An enumeration.

DAY = 3

FOK = 1

GTC = 2

class pyqstrat.pq_types.Trade(contract, order, timestamp, qty, price, fee=0.0, commission=0.0, properties=None)

Bases: object

__init__(contract, order, timestamp, qty, price, fee=0.0, commission=0.0, properties=None)

Parameters:

• contract (Contract) – The contract we traded

• order (Order) – A reference to the order that created this trade. Default None

• timestamp (datetime64) – Trade execution datetime

• qty (float) – Number of contracts or shares filled

• price (float) – Trade price

• fee (float) – Fees paid to brokers or others. Default 0

• commision – Commission paid to brokers or others. Default 0

• properties (Optional[SimpleNamespace]) – Any data you want to store with this contract. For example, you may want to store bid / ask prices at time of trade. Default None

class pyqstrat.pq_types.VWAPOrder(*, contract, timestamp=numpy.datetime64('NaT'), qty=nan, reason_code='', time_in_force=TimeInForce.FOK, properties=<factory>, status=OrderStatus.OPEN, vwap_stop=nan, vwap_end_time)

Bases: Order

An order type to trade at VWAP. A vwap order executes at VWAP from the point it is sent to the market till the vwap end time specified in the order.

Parameters:

• vwap_stop (float) – limit price. If market price <= vwap_stop for buys or market price

• sells (>= vwap_stop for) –

• point. (the order is executed at that) –

• vwap_end_time (datetime64) – We want to execute at VWAP computed from now to this time

vwap_end_time: datetime64

vwap_stop: float = nan

pyqstrat.pq_io module

pyqstrat.pq_io.df_to_hdf5(df, filename, key, dtypes=None, as_utf8=None)

Write out a pandas dataframe to hdf5 using the np_arrays_to_hdf5 function

Return type:

None

pyqstrat.pq_io.hdf5_copy(in_filename, in_key, out_filename, out_key=None, skip_if_exists=True)

Recursively copy groups from input filename to output filename. :type skip_if_exists: bool :param skip_if_exists: if set, we will skip any groups in the output that already exist. :param Otherwise we replace them.:

>>> tempdir = get_temp_dir()
>>> in_filename = tempdir + '/temp_in.hdf5'
>>> out_filename = tempdir + '/temp_out.hdf5'
>>> for filename in [in_filename, out_filename]:
...    if os.path.exists(filename): os.remove(filename)
>>> key = "test_g/test_subg"
>>> with h5py.File(in_filename, 'w') as f:
...    g = f.create_group(key)
...    ds = g.create_dataset('a', data = np.array([5, 3, 2.]))
>>> hdf5_copy(in_filename, key, out_filename, key)
>>> with h5py.File(out_filename, 'r') as f:
...    assert_(key in f)
>>> hdf5_copy(in_filename, key, out_filename, key)
>>> with h5py.File(out_filename, 'r') as f:
...    assert_(key in f)
>>> hdf5_copy(in_filename, key, out_filename, key, False)
:rtype: :py:obj:`None`

>>> with h5py.File(out_filename, 'r') as f:
...    assert_(key in f)

pyqstrat.pq_io.hdf5_repack(in_filename, out_filename)

Copy groups from input filename to output filename. Serves the same purpose as the h5repack command line tool, i.e. discards empty space in the input file so the output file may be smaller

Return type:

None

pyqstrat.pq_io.hdf5_to_df(filename, key)

Read a pandas dataframe previously written out using df_to_hdf5 or np_arrays_to_hdf5

Return type:

DataFrame

pyqstrat.pq_io.hdf5_to_np_arrays(filename, key)

Read a list of numpy arrays previously written out by np_arrays_to_hdf5 :type filename: str :param filename: path of the hdf5 file to read :type key: str :param key: group and or / subgroups to read from. For example, “g1/g2” will read from the subgrp g2 within the grp g1

Return type:

dict[str, ndarray]

Returns:

a list of numpy arrays along with their names

pyqstrat.pq_io.np_arrays_to_hdf5(data, filename, key, dtypes=None, as_utf8=None, compression_args=None)

Write a list of numpy arrays to hdf5 :type data: dict[str, ndarray] :param data: list of numpy one dimensional arrays along with the name of the array :type filename: str :param filename: filename of the hdf5 file :type key: str :param key: group and or / subgroups to write to. For example, “g1/g2” will write to the subgrp g2 within the grp g1 :type dtypes: Optional[dict[str, str]] :param dtypes: dict used to override datatype for a column. For example, {“col1”: “f4”} will write a 4 byte float array for col1 :param as_utf_8: each column listed here will be saved with utf8 encoding. For all other strings, we will compute the max length :param and store as a fixed length byte array with ascii encoding: :param i.e. a S[max length] datatype. This is much faster to read and process: :type compression_args: Optional[dict[Any, Any]] :param compression_args: if you want to compress the hdf5 file. You can use the hdf5plugin module and arguments such as hdf5plugin.Blosc()

Return type:

None

pyqstrat.pq_io.test_hdf5_to_df()

pyqstrat.holiday_calendars module

class pyqstrat.holiday_calendars.Calendar(calendar_name)

Bases: object

__init__(calendar_name)

Create a calendar object :type calendar_name: str :param calendar_name: name of calendar as defined in the pandas_market_calendars package :type calendar_name: str

add_trading_days(start, num_days, roll='raise')

Adds trading days to a start date

Parameters:

• start (Union[Timestamp, str, datetime64, datetime, date]) – start datetimes(s)

• num_days (int | ndarray) – number of trading days to add

• roll (str) – one of ‘raise’, ‘nat’, ‘forward’, ‘following’, ‘backward’, ‘preceding’, ‘modifiedfollowing’, ‘modifiedpreceding’ or ‘allow’} ‘allow’ is a special case in which case, adding 1 day to a holiday will act as if it was not a holiday, and give you the next business day’ The rest of the values are the same as in the numpy busday_offset function From numpy documentation: How to treat dates that do not fall on a valid day. The default is ‘raise’. ‘raise’ means to raise an exception for an invalid day. ‘nat’ means to return a NaT (not-a-time) for an invalid day. ‘forward’ and ‘following’ mean to take the first valid day later in time. ‘backward’ and ‘preceding’ mean to take the first valid day earlier in time. ‘modifiedfollowing’ means to take the first valid day later in time unless it is across a Month boundary, in which case to take the first valid day earlier in time. ‘modifiedpreceding’ means to take the first valid day earlier in time unless it is across a Month boundary, in which case to take the first valid day later in time.

Return type:

datetime64 | ndarray

Returns:

The datetime num_days trading days after start

>>> calendar = Calendar('NYSE')
>>> calendar.add_trading_days(datetime.date(2015, 12, 24), 1)
numpy.datetime64('2015-12-28')
>>> calendar.add_trading_days(np.datetime64('2017-04-15'), 0, roll = 'preceding') # 4/14/2017 is a Friday and a holiday
numpy.datetime64('2017-04-13')
>>> calendar.add_trading_days(np.datetime64('2017-04-08'), 0, roll = 'preceding') # 4/7/2017 is a Friday and not a holiday
numpy.datetime64('2017-04-07')
>>> calendar.add_trading_days(np.datetime64('2019-02-17 15:25'), 1, roll = 'allow')
numpy.datetime64('2019-02-19T15:25')
>>> calendar.add_trading_days(np.datetime64('2019-02-17 15:25'), -1, roll = 'allow')
numpy.datetime64('2019-02-15T15:25')

get_trading_days(start, end, include_first=False, include_last=True)

Get back a list of numpy dates that are trading days between the start and end

>>> nyse = Calendar('NYSE')
>>> nyse.get_trading_days('2005-01-01', '2005-01-08')
array(['2005-01-03', '2005-01-04', '2005-01-05', '2005-01-06', '2005-01-07'], dtype='datetime64[D]')
>>> nyse.get_trading_days(datetime.date(2005, 1, 1), datetime.date(2005, 2, 1))
array(['2005-01-03', '2005-01-04', '2005-01-05', '2005-01-06',
       '2005-01-07', '2005-01-10', '2005-01-11', '2005-01-12',
       '2005-01-13', '2005-01-14', '2005-01-18', '2005-01-19',
       '2005-01-20', '2005-01-21', '2005-01-24', '2005-01-25',
       '2005-01-26', '2005-01-27', '2005-01-28', '2005-01-31', '2005-02-01'], dtype='datetime64[D]')
>>> nyse.get_trading_days(datetime.date(2016, 1, 5), datetime.date(2016, 1, 29), include_last = False)
array(['2016-01-06', '2016-01-07', '2016-01-08', '2016-01-11',
       '2016-01-12', '2016-01-13', '2016-01-14', '2016-01-15',
       '2016-01-19', '2016-01-20', '2016-01-21', '2016-01-22',
       '2016-01-25', '2016-01-26', '2016-01-27', '2016-01-28'], dtype='datetime64[D]')
>>> nyse.get_trading_days('2017-07-04', '2017-07-08', include_first = False)
array(['2017-07-05', '2017-07-06', '2017-07-07'], dtype='datetime64[D]')
:rtype: :py:class:`int` | :py:class:`~numpy.ndarray`

>>> nyse.get_trading_days(np.datetime64('2017-07-04'), np.datetime64('2017-07-08'), include_first = False)
array(['2017-07-05', '2017-07-06', '2017-07-07'], dtype='datetime64[D]')

is_trading_day(dates)

Returns whether the date is not a holiday or a weekend

Parameters:

dates (Union[Timestamp, str, datetime64, datetime, date]) – date times to check

Return type:

bool | ndarray

Returns:

Whether this date is a trading day

>>> import datetime
>>> eurex = Calendar('EUREX')
>>> eurex.is_trading_day('2016-12-25')
False
>>> eurex.is_trading_day(datetime.date(2016, 12, 22))
True
>>> nyse = Calendar('NYSE')
>>> nyse.is_trading_day('2017-04-01') # Weekend
False
>>> nyse.is_trading_day(np.arange('2017-04-01', '2017-04-09', dtype = np.datetime64)) 
array([False, False,  True,  True,  True,  True,  True, False]...)

num_trading_days(start, end, include_first=False, include_last=True)

Count the number of trading days between two date series including those two dates

>>> eurex = Calendar('EUREX')
>>> eurex.num_trading_days('2009-01-01', '2011-12-31')
766.0
>>> dates = np.arange(np.datetime64('2013-01-01'),np.datetime64('2013-01-09'), np.timedelta64(1, 'D'))
>>> increments = np.array([5, 0, 3, 9, 4, 10, 15, 29])
>>> import warnings
>>> import pandas as pd
>>> warnings.filterwarnings(action = 'ignore', category = pd.errors.PerformanceWarning)
>>> dates2 = dates + increments
>>> dates[4] = np.datetime64('NaT')
>>> dates2[6] = np.datetime64('NaT')
>>> df = pd.DataFrame({'x': dates, 'y' : dates2})
>>> nyse = Calendar('NYSE')
>>> np.set_printoptions(formatter = {'float' : lambda x : f'{x:.1f}'})  # After numpy 1.13 positive floats don't have a leading space for sign
:rtype: :py:class:`float` | :py:class:`~numpy.ndarray`

>>> print(nyse.num_trading_days(df.x, df.y))
[3.0 0.0 1.0 5.0 nan 8.0 nan 20.0]

third_friday_of_month(month, year, roll='backward')

>>> nyse = Calendar('NYSE')
:rtype: :py:class:`~numpy.datetime64`

>>> nyse.third_friday_of_month(3, 2017)
numpy.datetime64('2017-03-17')

pyqstrat.holiday_calendars.get_date_from_weekday(weekday, year, month, week)

Return the date that falls on a given weekday (Monday = 0) on a week, year and month >>> get_date_from_weekday(1, 2019, 10, 4) numpy.datetime64(‘2019-10-22’)

Return type:

datetime64

pyqstrat.account module

class pyqstrat.account.Account(contract_groups, timestamps, price_function, strategy_context, starting_equity=1000000.0, pnl_calc_time=900)

Bases: object

An Account calculates pnl for a set of contracts

__init__(contract_groups, timestamps, price_function, strategy_context, starting_equity=1000000.0, pnl_calc_time=900)

Parameters:

• contract_groups (Sequence[ContractGroup]) – Contract groups that we want to compute PNL for

• timestamps (ndarray) – Timestamps that we might compute PNL at

• price_function (Callable[[Contract, ndarray, int, SimpleNamespace], float]) – Function that returns contract prices used to compute pnl

• strategy_context (SimpleNamespace) – This is passed into the price function so we can use current state of strategy to compute prices

• starting_equity (float) – Starting equity in account currency. Default 1.e6

• pnl_calc_time (int) – Number of minutes past midnight that we should calculate PNL at. Default 15 * 60, i.e. 3 pm

add_trades(trades)

Return type:

None

calc(timestamp)

Computes P&L and stores it internally for all contracts.

Parameters:

timestamp (datetime64) – timestamp to compute P&L at. Account remembers the last timestamp it computed P&L up to and will compute P&L between these and including timestamp. If there is more than one day between the last index and current index, we will include pnl for at the defined pnl_calc_time for those dates as well.

Return type:

None

df_account_pnl(contract_group=None)

Returns PNL at the account level.

Parameters:

contract_group (Optional[ContractGroup]) – If set, we only return pnl for this contract_group. Otherwise we return pnl for all contract groups

Return type:

DataFrame

df_pnl(contract_groups=None)

Returns a dataframe with P&L columns broken down by contract group and symbol

Parameters:

contract_group – Return PNL for this contract group. If None (default), include all contract groups

Return type:

DataFrame

df_roundtrip_trades(contract_group=None, start_date=numpy.datetime64('NaT'), end_date=numpy.datetime64('NaT'))

Returns a dataframe of round trip trades

Parameters:

• contract_group (Optional[ContractGroup]) – Return trades for this contract group. If None (default), include all contract groups

• start_date (datetime64) – Include trades with date greater than or equal to this timestamp.

• end_date (datetime64) – Include trades with date less than or equal to this timestamp.

Return type:

DataFrame

df_trades(contract_group=None, start_date=numpy.datetime64('NaT'), end_date=numpy.datetime64('NaT'))

Returns a dataframe of trades

Parameters:

• contract_group (Optional[ContractGroup]) – Return trades for this contract group. If None (default), include all contract groups

• start_date (datetime64) – Include trades with date greater than or equal to this timestamp.

• end_date (datetime64) – Include trades with date less than or equal to this timestamp.

Return type:

DataFrame

equity(timestamp)

Returns equity in this account in Account currency. Will cause calculation if Account has not previously calculated up to this date

Return type:

float

get_trades_for_date(symbol, date)

Return type:

list[Trade]

position(contract_group, timestamp)

Returns netted position for a contract_group at a given date in number of contracts or shares.

Return type:

float

positions(contract_group, timestamp)

Returns all non-zero positions in a contract group

Return type:

list[tuple[Contract, float]]

roundtrip_trades(contract_group=None, start_date=numpy.datetime64('NaT'), end_date=numpy.datetime64('NaT'))

Returns a list of round trip trades with the given symbol and with trade date between (and including) start date and end date if they are specified. If symbol is None trades for all symbols are returned

Return type:

list[RoundTripTrade]

symbols()

Return type:

list[str]

trades(contract_group=None, start_date=numpy.datetime64('NaT'), end_date=numpy.datetime64('NaT'))

Returns a list of trades with the given symbol and with trade date between (and including) start date and end date if they are specified. If symbol is None trades for all symbols are returned

Return type:

list[Trade]

class pyqstrat.account.ContractPNL(contract, account_timestamps, price_function, strategy_context)

Bases: object

Computes pnl for a single contract over time given trades and market data >>> from pyqstrat.pq_types import MarketOrder >>> Contract.clear_cache() >>> aapl_contract = Contract.create(‘AAPL’) >>> timestamps = np.arange(np.datetime64(‘2018-01-01’), np.datetime64(‘2018-01-04’)) >>> def get_price(contract, timestamps, idx, strategy_context): … assert contract.symbol == ‘AAPL’, f’unknown contract: {contract}’ … return idx + 10.1

>>> contract_pnl = ContractPNL(aapl_contract, timestamps, get_price, SimpleNamespace())
>>> trade_5 = Trade(aapl_contract, MarketOrder(contract=aapl_contract, timestamp=timestamps[1], qty=20), timestamps[2], 10, 16.2)
>>> trade_6 = Trade(aapl_contract, MarketOrder(contract=aapl_contract, timestamp=timestamps[1], qty=-20), timestamps[2], -10, 16.5)
>>> trade_7 = Trade(aapl_contract, MarketOrder(contract=aapl_contract, timestamp=timestamps[1], qty=-20), timestamps[2], -10, 16.5)
>>> contract_pnl._add_trades([trade_5, trade_6])
>>> contract_pnl._add_trades([trade_7])
>>> df = contract_pnl.df()
>>> assert (len(df == 1))
>>> row = df.iloc[0]
>>> assert row.to_dict() == {'symbol': 'AAPL',
... 'timestamp': pd.Timestamp('2018-01-03 00:00:00'),
... 'position': -10,
... 'price': 12.1,
... 'unrealized': 44.0,
... 'realized': 3.000000000000007,
... 'commission': 0.0,
... 'fee': 0.0,
... 'net_pnl': 47.00000000000001}

calc_net_pnl(timestamp)

Return type:

None

df()

Returns a pandas dataframe with pnl data

Return type:

DataFrame

net_pnl(timestamp)

Return type:

float

pnl(timestamp)

Return type:

tuple[float, float, float, float, float, float, float]

position(timestamp)

Return type:

float

pyqstrat.account.find_index_before(sorted_dict, key)

Find index of the first key in a sorted dict that is less than or equal to the key passed in. If the key is less than the first key in the dict, return -1

Return type:

int

pyqstrat.account.find_last_non_nan_index(array)

Return type:

int

pyqstrat.account.leading_nan_to_zero(df, columns)

Return type:

DataFrame

pyqstrat.account.test_account()

pyqstrat.strategy module

class pyqstrat.strategy.Strategy(timestamps, contract_groups, price_function, starting_equity=1000000.0, pnl_calc_time=961, trade_lag=0, run_final_calc=True, log_trades=True, log_orders=False, strategy_context=None)

Bases: object

__init__(timestamps, contract_groups, price_function, starting_equity=1000000.0, pnl_calc_time=961, trade_lag=0, run_final_calc=True, log_trades=True, log_orders=False, strategy_context=None)

Parameters:

• timestamps (np.array of np.datetime64) – The “heartbeat” of the strategy. We will evaluate trading rules and simulate the market at these times.

• contract_groups (Sequence[ContractGroup]) – The contract groups we will potentially trade.

• price_function (Callable[[Contract, ndarray, int, SimpleNamespace], float]) – A function that returns the price of a contract at a given timestamp

• starting_equity (float) – Starting equity in Strategy currency. Default 1.e6

• pnl_calc_time (int) – Time of day used to calculate PNL. Default 15 * 60 (3 pm)

• trade_lag (int) – Number of bars you want between the order and the trade. For example, if you think it will take 5 seconds to place your order in the market, and your bar size is 1 second, set this to 5. Set this to 0 if you want to execute your trade at the same time as you place the order, for example, if you have daily bars. Default 0.

• run_final_calc (bool) – If set, calculates unrealized pnl and net pnl as well as realized pnl when strategy is done. If you don’t need unrealized pnl, turn this off for faster run time. Default True

• strategy_context (Optional[SimpleNamespace]) – A storage class where you can store key / value pairs relevant to this strategy. For example, you may have a pre-computed table of correlations that you use in the indicator or trade rule functions. If not set, the __init__ function will create an empty member strategy_context object that you can access.

• log_trades (bool) – If set, we log orders as they are created

• log_orders (bool) – If set, we log trades as they are created

add_indicator(name, indicator, contract_groups=None, depends_on=None)

Parameters:

• name (str) – Name of the indicator

• indicator (Callable[[ContractGroup, ndarray, SimpleNamespace, SimpleNamespace], ndarray]) – A function that takes strategy timestamps and other indicators and returns a numpy array containing indicator values. The return array must have the same length as the timestamps object.

• contract_groups (Optional[Sequence[ContractGroup]]) – Contract groups that this indicator applies to. If not set, it applies to all contract groups. Default None.

• depends_on (Optional[Sequence[str]]) – Names of other indicators that we need to compute this indicator. Default None.

Return type:

None

add_market_sim(market_sim_function)

Add a market simulator. A market simulator is a function that takes orders as input and returns trades.

Return type:

None

add_rule(name, rule_function, signal_name, sig_true_values=None, position_filter=None)

Add a trading rule. Trading rules are guaranteed to run in the order in which you add them. For example, if you set trade_lag to 0,

and want to exit positions and re-enter new ones in the same bar, make sure you add the exit rule before you add the entry rule to the strategy.

Parameters:

• name (str) – Name of the trading rule

• rule_function (Callable[[ContractGroup, int, ndarray, SimpleNamespace, ndarray, Account, Sequence[Order], SimpleNamespace], list[Order]]) – A trading rule function that returns a list of Orders

• signal_name (str) – The strategy will call the trading rule function when the signal with this name matches sig_true_values

• sig_true_values (Optional[Sequence[Any]]) – If the signal value at a bar is equal to one of these values, the Strategy will call the trading rule function. Default [TRUE]

• position_filter (Optional[str]) – Can be “zero”, “nonzero”, “positive”, “negative” or None. Rules are only triggered when the corresponding contract positions fit the criteria. For example, a positive rule is only triggered when the current position for that contract is > 0 If not set, we don’t look at the position before triggering the rule. Default None

Return type:

None

add_signal(name, signal_function, contract_groups=None, depends_on_indicators=None, depends_on_signals=None)

Parameters:

• name (str) – Name of the signal

• signal_function (function) – A function that takes timestamps and a dictionary of indicator value arrays and returns a numpy array containing signal values. The return array must have the same length as the input timestamps

• contract_groups (list of ContractGroup, optional) – Contract groups that this signal applies to. If not set, it applies to all contract groups. Default None.

• depends_on_indicators (list of str, optional) – Names of indicators that we need to compute this signal. Default None.

• depends_on_signals (list of str, optional) – Names of other signals that we need to compute this signal. Default None.

Return type:

None

df_data(contract_groups=None, add_pnl=True, start_date=numpy.datetime64('NaT'), end_date=numpy.datetime64('NaT'))

Add indicators and signals to end of market data and return as a pandas dataframe.

Parameters:

• contract_groups (list of:obj:ContractGroup, optional) – list of contract groups to include. All if set to None (default)

• add_pnl (bool) – If True (default), include P&L columns in dataframe

• start_date (str | datetime64) – string or numpy datetime64. Default None

• end_date (str | datetime64) – string or numpy datetime64: Default None

Return type:

DataFrame

df_orders(contract_group=None, start_date=numpy.datetime64('NaT'), end_date=numpy.datetime64('NaT'))

Returns a dataframe with data from orders with the given contract group and with order date between (and including) start date and end date if they are specified. If contract_group is None orders for all contract_groups are returned

Return type:

DataFrame

df_pnl(contract_group=None)

Returns a dataframe with P&L columns. If contract group is set to None (default), sums up P&L across all contract groups

Return type:

DataFrame

df_returns(contract_group=None, sampling_frequency='D')

Return a dataframe of returns and equity indexed by date.

Parameters:

• contract_group (Optional[ContractGroup]) – The contract group to get returns for. If set to None (default), we return the sum of PNL for all contract groups

• sampling_frequency (str) – Downsampling frequency. Default is None. See pandas frequency strings for possible values

Return type:

DataFrame

df_roundtrip_trades(contract_group=None, start_date=numpy.datetime64('NaT'), end_date=numpy.datetime64('NaT'))

Returns a dataframe of round trip trades with the given contract group and with trade date between (and including) start date and end date if they are specified. If contract_group is None trades for all contract_groups are returned

Return type:

DataFrame

df_trades(contract_group=None, start_date=numpy.datetime64('NaT'), end_date=numpy.datetime64('NaT'))

Returns a dataframe with data from trades with the given contract group and with trade date between (and including) start date and end date if they are specified. If contract_group is None trades for all contract_groups are returned

Return type:

DataFrame

evaluate_returns(contract_group=None, periods_per_year=0, plot=True, display_summary=True, float_precision=4, return_metrics=True)

Computes return metrics and does or more of plotting, displaying or returning them.

Parameters:

• contract_group (ContractGroup, optional) – Contract group to evaluate or None (default) for all contract groups

• periods_per_year (int) – If set to 0, we try to infer the frequency from the timestamps in the returns sometimes this is not possible, for example if you have daily returns with random gaps in the days In that case, you should set this value yourself. Use 252 if returns are on a daily frequency

• plot (bool) – If set to True, display plots of equity, drawdowns and returns. Default False

• float_precision (float, optional) – Number of significant figures to show in returns. Default 4

• return_metrics (bool, optional) – If set, we return the computed metrics as a dictionary

Return type:

Optional[dict[str, Any]]

orders(contract_group=None, start_date=None, end_date=None)

Returns a list of orders with the given contract group and with order date between (and including) start date and end date if they are specified. If contract_group is None orders for all contract_groups are returned

Return type:

list[Order]

plot_returns(contract_group=None)

Display plots of equity, drawdowns and returns for the given contract group or for all contract groups if contract_group is None (default)

Return type:

Figure

roundtrip_trades(contract_group=None, start_date=numpy.datetime64('NaT'), end_date=numpy.datetime64('NaT'))

Returns a list of trades with the given contract group and with trade date between (and including) start date and end date if they are specified. If contract_group is None trades for all contract_groups are returned

Return type:

list[RoundTripTrade]

run()

Return type:

None

run_indicators(indicator_names=None, contract_groups=None, clear_all=False)

Calculate values of the indicators specified and store them.

Parameters:

• indicator_names (Optional[Sequence[str]]) – list of indicator names. If None (default) run all indicators

• contract_groups (Optional[Sequence[ContractGroup]]) – Contract group to run this indicator for. If None (default), we run it for all contract groups.

• clear_all (bool) – If set, clears all indicator values before running. Default False.

Return type:

None

run_rules(rule_names=None, contract_groups=None, start_date=numpy.datetime64('NaT'), end_date=numpy.datetime64('NaT'))

Run trading rules.

Parameters:

• rule_names (Optional[Sequence[str]]) – list of rule names. If None (default) run all rules

• contract_groups (Optional[Sequence[ContractGroup]]) – Contract groups to run this rule for. If None (default), we run it for all contract groups.

• start_date (datetime64) – Run rules starting from this date. Default None

• end_date (datetime64) – Don’t run rules after this date. Default None

Return type:

None

run_signals(signal_names=None, contract_groups=None, clear_all=False)

Calculate values of the signals specified and store them.

Parameters:

• signal_names (Optional[Sequence[str]]) – list of signal names. If None (default) run all signals

• contract_groups (Optional[Sequence[ContractGroup]]) – Contract groups to run this signal for. If None (default), we run it for all contract groups.

• clear_all (bool) – If set, clears all signal values before running. Default False.

Return type:

None

trades(contract_group=None, start_date=numpy.datetime64('NaT'), end_date=numpy.datetime64('NaT'))

Returns a list of trades with the given contract group and with trade date between (and including) start date and end date if they are specified. If contract_group is None trades for all contract_groups are returned

Return type:

list[Trade]

pyqstrat.portfolio module

class pyqstrat.portfolio.Portfolio(name='main')

Bases: object

A portfolio contains one or more strategies that run concurrently so you can test running strategies that are uncorrelated together.

__init__(name='main')

Args: name: String used for displaying this portfolio

add_strategy(name, strategy)

Parameters:

• name (str) – Name of the strategy

• strategy (Strategy) – Strategy instance

Return type:

None

df_returns(sampling_frequency='D', strategy_names=None)

Return dataframe containing equity and returns with a date index. Equity and returns are combined from all strategies passed in.

Parameters:

• sampling_frequency (str) – Date frequency for rows. Default ‘D’ for daily so we will have one row per day

• strategy_names (Optional[Sequence[str]]) – By default this is set to None and we use all strategies.

Return type:

DataFrame

evaluate_returns(sampling_frequency='D', strategy_names=None, plot=True, float_precision=4)

Returns a dictionary of common return metrics.

Parameters:

• sampling_frequency (str) – Date frequency. Default ‘D’ for daily so we downsample to daily returns before computing metrics

• strategy_names (Optional[Sequence[str]]) – By default this is set to None and we use all strategies.

• plot (bool) – If set to True, display plots of equity, drawdowns and returns. Default False

• float_precision (int) – Number of significant figures to show in returns. Default 4

Return type:

dict[Any, Any]

plot(sampling_frequency='D', strategy_names=None)

Display plots of equity, drawdowns and returns

Parameters:

• sampling_frequency (str) – Date frequency. Default ‘D’ for daily so we downsample to daily returns before computing metrics

• strategy_names (Optional[Sequence[str]]) – A list of strategy names. By default this is set to None and we use all strategies.

Return type:

None

run(strategy_names=None, start_date=numpy.datetime64('NaT'), end_date=numpy.datetime64('NaT'))

Run indicators, signals and rules.

Parameters:

• strategy_names (Optional[Sequence[str]]) – A list of strategy names. By default this is set to None and we use all strategies.

• start_date (datetime64) – Run rules starting from this date. Sometimes we have a few strategies in a portfolio that need different lead times before they are ready to trade so you can set this so they are all ready by this date. Default None

• end_date (datetime64) – Don’t run rules after this date. Default None

Return type:

None

run_indicators(strategy_names=None)

Compute indicators for the strategies specified

Parameters:

strategy_names (Optional[Sequence[str]]) – By default this is set to None and we use all strategies.

Return type:

None

run_rules(strategy_names=None, start_date=numpy.datetime64('NaT'), end_date=numpy.datetime64('NaT'))

Run rules for the strategies specified. Must be called after run_indicators and run_signals. See run function for argument descriptions

Return type:

None

run_signals(strategy_names=None)

Compute signals for the strategies specified. Must be called after run_indicators

Parameters:

strategy_names (Optional[Sequence[str]]) – By default this is set to None and we use all strategies.

Return type:

None

pyqstrat.optimize module

class pyqstrat.optimize.Experiment(suggestion, cost, other_costs)

Bases: object

An Experiment stores a suggestion and its result

__init__(suggestion, cost, other_costs)

Parameters:

• suggestion (dict[str, Any]) – A dictionary of variable name -> value

• cost (float) – A float representing output of the function we are testing with this suggestion as input.

• other_costs (dict[str, float]) – A dictionary of other results we want to store and look at later.

valid()

Returns True if all suggestions and costs are finite, i.e not NaN or +/- Infinity

Return type:

bool

class pyqstrat.optimize.Optimizer(name, generator, cost_func, max_processes=None)

Bases: object

Optimizer is used to optimize parameters for a strategy.

__init__(name, generator, cost_func, max_processes=None)

Parameters:

• name (str) – Display title for plotting, etc.

• generator (Generator[dict[str, Any], tuple[float, dict[str, float]], None]) – A generator (see Python Generators) that takes no inputs and yields a dictionary with parameter name -> parameter value.

• cost_func (Callable[[dict[str, Any]], tuple[float, dict[str, float]]]) – A function that takes a dictionary of parameter name -> parameter value as input and outputs cost for that set of parameters.

• max_processes (Optional[int]) – If not set, the Optimizer will look at the number of CPU cores on your machine to figure out how many processes to run.

df_experiments(sort_column='cost', ascending=True)

Returns a dataframe containing experiment data, sorted by sort_column (default “cost”)

Return type:

DataFrame

experiment_list(sort_order='lowest_cost')

Returns the list of experiments we have run

Parameters:

sort_order (str) – Can be set to lowest_cost, highest_cost or sequence. If set to sequence, experiments are returned in the sequence in which they were run

Return type:

Sequence[Experiment]

plot_2d(x, y='all', title='', marker_mode='lines+markers', height=1000, width=0, show=True)

Creates a 2D plot of the optimization output for plotting 1 parameter and costs

Parameters:

• x (str) – Name of the parameter to plot on the x axis, corresponding to the same name in the generator.

• y (str) – Can be one of: “cost” The name of another cost variable corresponding to the output from the cost function “all”, which creates a subplot for cost plus all other costs

• marker_mode (str) – see plotly mode. Set to ‘lines’ to turn markers off

Return type:

Figure

plot_3d(x, y, z='all', markers=True, filter_func=None, height=1000, width=0, xlim=None, ylim=None, vertical_spacing=0.05, show=True)

Creates a 3D plot of the optimization output for plotting 2 parameters and costs.

Parameters:

• x (str) – Name of the parameter to plot on the x axis, corresponding to the same name in the generator.

• y (str) – Name of the parameter to plot on the y axis, corresponding to the same name in the generator.

• z (str) – Can be one of cost, the name of another cost variable corresponding to the output from the cost function all, which creates a subplot for cost plus all other costs

• markers (bool) – If set, we show actual datapoints on the graph

• filter_func (Optional[Callable[[DataFrame], DataFrame]]) – A function that can be used to reduce the dataset before plotting. For example, you may want to filter on a dimension beyond x, y, z to pick a single value from that dimension

• marker – Adds a marker to each point in x, y, z to show the actual data used for interpolation. You can set this to None to turn markers off.

• vertical_spacing (float) – Vertical space between subplots

Return type:

Figure

run(raise_on_error=False)

Run the optimizer.

Parameters:

raise_on_error (bool) – If set to True, even if we are running a multiprocess optimization, any Exceptions will bubble up and stop the Optimizer. This can be useful for debugging to see stack traces for Exceptions.

Return type:

None

pyqstrat.optimize.flatten_keys(experiments)

Utility function so we can find all keys for other costs in all experiments even if the first experiment does not have all of them

Return type:

list[str]

pyqstrat.optimize.test_optimize()

pyqstrat.evaluator module

class pyqstrat.evaluator.Evaluator(initial_metrics)

Bases: object

You add functions to the evaluator that are dependent on the outputs of other functions. The evaluator will call these functions in the right order so dependencies are computed first before the functions that need their output. You can retrieve the output of a metric using the metric member function

>>> evaluator = Evaluator(initial_metrics={'x': np.array([1, 2, 3]), 'y': np.array([3, 4, 5])})
>>> evaluator.add_metric('z', lambda x, y: sum(x, y), dependencies=['x', 'y'])
>>> evaluator.compute()
>>> evaluator.metric('z')
array([ 9, 10, 11])

__init__(initial_metrics)

Inits Evaluator with a dictionary of initial metrics that are used to compute subsequent metrics

Parameters:

initial_metrics (dict[str, Any]) – a dictionary of string name -> metric. metric can be any object including a scalar, an array or a tuple

add_metric(name, func, dependencies)

Return type:

None

compute(metric_names=None)

Compute metrics using the internal dependency graph

Parameters:

metric_names (Optional[Sequence[str]]) – an array of metric names. If not passed in, evaluator will compute and store all metrics

Return type:

None

compute_metric(metric_name)

Compute and store a single metric:

Parameters:

metric_name (str) – string representing the metric to compute

Return type:

None

metric(metric_name)

Return the value of a single metric given its name

Return type:

Any

metrics()

Return a dictionary of metric name -> metric value

Return type:

dict[str, Any]

pyqstrat.evaluator.compute_amean(returns, periods_per_year)

Computes arithmetic mean of a return array, ignoring NaNs

Parameters:

• returns (ndarray) – Represents returns at any frequency

• periods_per_year (int) – Frequency of the returns, e.g. 252 for daily returns

Return type:

float

>>> compute_amean(np.array([0.003, 0.004, np.nan]), 252)
0.882

pyqstrat.evaluator.compute_annual_returns(timestamps, returns, periods_per_year)

Takes the output of compute_bucketed_returns and returns geometric mean of returns by year

Return type:

tuple[ndarray, ndarray]

Returns:

A tuple with the first element being an array of years (integer) and the second element an array of annualized returns for those years

pyqstrat.evaluator.compute_bucketed_returns(timestamps, returns)

Bucket returns by year

Return type:

tuple[list[int], list[ndarray]]

Returns:

A tuple with the first element being a list of years and the second a list of

numpy arrays containing returns for each corresponding year

pyqstrat.evaluator.compute_calmar(returns_3yr, periods_per_year, mdd_pct_3yr)

Compute Calmar ratio, which is the annualized return divided by max drawdown over the last 3 years

Return type:

float

pyqstrat.evaluator.compute_dates_3yr(timestamps)

Given an array of numpy datetimes, return those that are within 3 years of the last date in the array

Return type:

ndarray

pyqstrat.evaluator.compute_equity(timestamps, starting_equity, returns)

Given starting equity, timestamps and returns, create a numpy array of equity at each date

Return type:

ndarray

pyqstrat.evaluator.compute_gmean(timestamps, returns, periods_per_year)

Compute geometric mean of an array of returns

Parameters:

• returns (ndarray) – a numpy array of returns, can contain nans

• periods_per_year (float) – Used for annualizing returns

Return type:

float

>>> round(compute_gmean(np.array(['2015-01-01', '2015-03-01', '2015-05-01'], dtype='M8[D]'), np.array([0.001, 0.002, 0.003]), 252.), 6)
0.018362

pyqstrat.evaluator.compute_k_ratio(equity, periods_per_year, halflife_years=None)

Compute k-ratio (2013 or original versions by Lars Kestner). See https://papers.ssrn.com/sol3/papers.cfm?abstract_id=2230949 We also implement a modification that allows higher weighting for more recent returns.

Parameters:

• equity (ndarray) – a numpy array of the equity in your account

• periods_per_year (int) – 252 for daily values

• halflife_years (Optional[float]) – If set, we use weighted linear regression to give less weight to older returns. In this case, we compute the original k-ratio which does not use periods per year or number of observations If not set, we compute the 2013 version of the k-ratio which weights k-ratio by sqrt(periods_per_year) / nobs

Return type:

float

Returns:

weighted or unweighted k-ratio

>>> np.random.seed(0)
>>> t = np.arange(1000)
>>> ret = np.random.normal(loc = 0.0025, scale = 0.01, size = len(t))
>>> equity = (1 + ret).cumprod()
>>> assert(math.isclose(compute_k_ratio(equity, 252, None), 3.888, abs_tol=0.001))
>>> assert(math.isclose(compute_k_ratio(equity, 252, 0.5), 602.140, abs_tol=0.001))

pyqstrat.evaluator.compute_mar(returns, periods_per_year, mdd_pct)

Compute MAR ratio, which is annualized return divided by biggest drawdown since inception.

Return type:

float

pyqstrat.evaluator.compute_maxdd_date(rolling_dd_dates, rolling_dd)

Compute date of max drawdown given numpy array of timestamps, and corresponding rolling dd percentages

Return type:

datetime64

pyqstrat.evaluator.compute_maxdd_date_3yr(rolling_dd_3yr_timestamps, rolling_dd_3yr)

Compute max drawdown date over the last 3 years

Return type:

datetime64

pyqstrat.evaluator.compute_maxdd_pct(rolling_dd)

Compute max drawdown percentage given a numpy array of rolling drawdowns, ignoring NaNs

Return type:

float

pyqstrat.evaluator.compute_maxdd_pct_3yr(rolling_dd_3yr)

Compute max drawdown percentage over the last 3 years

Return type:

float

pyqstrat.evaluator.compute_maxdd_start(rolling_dd_dates, rolling_dd, mdd_date)

Compute date when max drawdown starts, given numpy array of timestamps corresponding rolling dd percentages and date of the max draw down

Return type:

datetime64

pyqstrat.evaluator.compute_maxdd_start_3yr(rolling_dd_3yr_timestamps, rolling_dd_3yr, mdd_date_3yr)

Comput max drawdown start date over the last 3 years

Return type:

datetime64

pyqstrat.evaluator.compute_num_periods(timestamps, periods_per_year)

Given an array of timestamps, we compute how many periods there are between the first and last element, where the length

of a period is defined by periods_per_year. For example, if there are 6 periods per year, then each period would be approx. 2 months long.

Parameters:

• timestamps (np.ndarray of np.datetime64) – a numpy array of returns, can contain nans

• periods_per_year (float) – number of periods between first and last return

Return type:

float

>>> assert(compute_num_periods(np.array(['2015-01-01', '2015-03-01', '2015-05-01', '2015-07-01'], dtype='M8[D]'), 6) == 3)

pyqstrat.evaluator.compute_periods_per_year(timestamps)

Computes trading periods per year for an array of numpy datetime64’s.

e.g. if most of the timestamps are separated by 1 day, will return 252.

Parameters:

timestamps (ndarray) – a numpy array of datetime64’s

>>> compute_periods_per_year(np.array(['2018-01-01', '2018-01-02', '2018-01-03', '2018-01-09', '2018-01-10'], dtype='M8[D]'))
:rtype: :py:class:`float`
252.0
>>> round(compute_periods_per_year(np.array(['2018-01-01 10:00', '2018-01-01 10:05', '2018-01-01 10:10'], dtype='M8[m]')), 2)
72576.0

pyqstrat.evaluator.compute_return_metrics(timestamps, rets, starting_equity, periods_per_year=0, leading_non_finite_to_zeros=False, subsequent_non_finite_to_zeros=True)

Compute a set of common metrics using returns (for example, of an instrument or a portfolio)

Parameters:

• timestamps (np.array of datetime64) – Timestamps for the returns

• rets (nd.array of float) – The returns, use 0.01 for 1%

• starting_equity (float) – Starting equity value in your portfolio

• leading_non_finite_to_zeros (bool, optional) – If set, we replace leading nan, inf, -inf returns with zeros. For example, you may need a warmup period for moving averages. Default False

• subsequent_non_finite_to_zeros (bool, optional) – If set, we replace any nans that follow the first non nan value with zeros. There may be periods where you have no prices but removing these returns would result in incorrect annualization. Default True

Return type:

Evaluator

Returns:

An Evaluator object containing computed metrics off the returns passed in. If needed, you can add your own metrics to this object based on the values of existing metrics and recompute the Evaluator. Otherwise, you can just use the output of the evaluator using the metrics function.

>>> timestamps = np.array(['2015-01-01', '2015-03-01', '2015-05-01', '2015-07-01'], dtype='M8[D]')
>>> rets = np.array([0.01, 0.02, np.nan, -0.015])
>>> starting_equity = 1.e6
>>> ev = compute_return_metrics(timestamps, rets, starting_equity)
>>> metrics = ev.metrics()
>>> assert(round(metrics['gmean'], 6) == 0.03122)
>>> assert(round(metrics['sharpe'], 6) == 0.594366)
>>> assert(all(metrics['returns_3yr'] == np.array([0.01, 0.02, 0, -0.015])))

pyqstrat.evaluator.compute_returns_3yr(timestamps, returns)

Given an array of numpy datetimes and an array of returns, return those that are within 3 years of the last date in the datetime array

Return type:

ndarray

pyqstrat.evaluator.compute_rolling_dd(timestamps, equity)

Compute numpy array of rolling drawdown percentage

Parameters:

• timestamps (ndarray) – numpy array of datetime64

• equity (ndarray) – numpy array of equity

Return type:

tuple[ndarray, ndarray]

pyqstrat.evaluator.compute_rolling_dd_3yr(timestamps, equity)

Compute rolling drawdowns over the last 3 years

Return type:

tuple[ndarray, ndarray]

pyqstrat.evaluator.compute_sharpe(returns, amean, periods_per_year)

Note that this does not take into risk free returns so it’s really a sharpe0, i.e. assumes risk free returns are 0

Parameters:

• returns (ndarray) – a numpy array of returns

• amean (float) – arithmetic mean of returns

• periods_per_year (float) – number of trading periods per year

Return type:

float

>>> round(compute_sharpe(np.array([0.001, -0.001, 0.002]), 0.001, 252), 6)
0.050508

pyqstrat.evaluator.compute_sortino(returns, amean, periods_per_year)

Note that this assumes target return is 0.

Parameters:

• returns (ndarray) – a numpy array of returns

• amean (float) – arithmetic mean of returns

• periods_per_year (float) – number of trading periods per year

Return type:

float

>>> print(round(compute_sortino(np.array([0.001, -0.001, 0.002]), 0.001, 252), 6))
0.133631

pyqstrat.evaluator.compute_std(returns)

Computes standard deviation of an array of returns, ignoring nans

Return type:

float

pyqstrat.evaluator.display_return_metrics(metrics, float_precision=3, show=True)

Creates a dataframe making it convenient to view the output of the metrics obtained using the compute_return_metrics function.

Parameters:

float_precision (int) – Change if you want to display floats with more or less significant figures than the default, 3 significant figures.

Return type:

DataFrame

Returns:

A one row dataframe with formatted metrics.

pyqstrat.evaluator.handle_non_finite_returns(timestamps, rets, leading_non_finite_to_zeros, subsequent_non_finite_to_zeros)

>>> np.set_printoptions(formatter={'float': '{: .6g}'.format})
>>> timestamps = np.arange(np.datetime64('2019-01-01'), np.datetime64('2019-01-07'))
>>> rets = np.array([np.nan, np.nan, 3, 4, np.nan, 5])
>>> handle_non_finite_returns(timestamps, rets, leading_non_finite_to_zeros = False, subsequent_non_finite_to_zeros = True)
(array(['2019-01-03', '2019-01-04', '2019-01-05', '2019-01-06'], dtype='datetime64[D]'), array([ 3,  4,  0,  5]))
>>> handle_non_finite_returns(timestamps, rets, leading_non_finite_to_zeros = True, subsequent_non_finite_to_zeros = False)
(array(['2019-01-01', '2019-01-02', '2019-01-03', '2019-01-04', '2019-01-06'], dtype='datetime64[D]'), array([ 0,  0,  3,  4,  5]))
>>> handle_non_finite_returns(timestamps, rets, leading_non_finite_to_zeros = False, subsequent_non_finite_to_zeros = False)
(array(['2019-01-01', '2019-01-02', '2019-01-03', '2019-01-04', '2019-01-06'], dtype='datetime64[D]'), array([ 0,  0,  3,  4,  5]))
>>> rets = np.array([1, 2, 3, 4, 4.5,  5])
>>> handle_non_finite_returns(timestamps, rets, leading_non_finite_to_zeros = False, subsequent_non_finite_to_zeros = True)
:rtype: :py:class:`tuple`\[:py:class:`~numpy.ndarray`, :py:class:`~numpy.ndarray`]

(array([‘2019-01-01’, ‘2019-01-02’, ‘2019-01-03’, ‘2019-01-04’, ‘2019-01-05’, ‘2019-01-06’],

dtype=’datetime64[D]’), array([ 1, 2, 3, 4, 4.5, 5]))

pyqstrat.evaluator.plot_return_metrics(metrics, title='', height=1000, width=0, show_points=False, show=True)

Plot equity, rolling drawdowns and and a boxplot of annual returns given the output of compute_return_metrics.

Parameters:

• metrics (dict[str, Any]) – dict of metrics produced by compute_return_metrics

• title – title of the plot

• height – height of the plot

• width – width of the plot. If set to 0, lets plotly select the width

• show – whether to show the plot or just return it

Return type:

Figure

pyqstrat.evaluator.test_evaluator()

Return type:

None

pyqstrat.pyqstrat_cpp module

pyqstrat.pyqstrat_cpp.black_scholes_price(call: numpy.ndarray[bool], S: numpy.ndarray[numpy.float64], K: numpy.ndarray[numpy.float64], t: numpy.ndarray[numpy.float64], r: numpy.ndarray[numpy.float64], sigma: numpy.ndarray[numpy.float64], q: numpy.ndarray[numpy.float64] = 0.0) → object

Compute Euroepean option price :param call: True for a call option, False for a put :type call: bool :param S: Spot price. For a future discount the future price using exp(-rt) :type S: float :param K: Strike :type K: float :param t: Time to maturity in years :type t: float :param r: Continuously compounded interest rate. Use 0.01 for 1% :type r: float :param sigma: Annualized volatility. Use 0.01 for 1% :type sigma: float :param q: Annualized dividend yield. Use 0.01 for 1%. Default 0 :type q: float, optional

Returns:

Option price

Return type:

float

pyqstrat.pyqstrat_cpp.cdf(x: numpy.ndarray[numpy.float64]) → object

Cumulative density function of normal distribution :param x: random variable :type x: float

Returns:

cdf of the random variable

Return type:

float

pyqstrat.pyqstrat_cpp.d1(S: numpy.ndarray[numpy.float64], K: numpy.ndarray[numpy.float64], t: numpy.ndarray[numpy.float64], r: numpy.ndarray[numpy.float64], sigma: numpy.ndarray[numpy.float64], q: numpy.ndarray[numpy.float64] = 0.0) → object

d1 from Black Scholes :param S: Spot price. For a future discount the future price using exp(-rt) :type S: float :param K: Strike :type K: float :param t: Time to maturity in years :type t: float :param r: Continuously compounded interest rate. Use 0.01 for 1% :type r: float :param sigma: Annualized volatility. Use 0.01 for 1% :type sigma: float :param q: Annualized dividend yield. Use 0.01 for 1% :type q: float

Return type:

float

pyqstrat.pyqstrat_cpp.d2(S: numpy.ndarray[numpy.float64], K: numpy.ndarray[numpy.float64], t: numpy.ndarray[numpy.float64], r: numpy.ndarray[numpy.float64], sigma: numpy.ndarray[numpy.float64], q: numpy.ndarray[numpy.float64] = 0.0) → object

d2 from Black Scholes :param S: Spot price. For a future discount the future price using exp(-rt) :type S: float :param K: Strike :type K: float :param t: Time to maturity in years :type t: float :param r: Continuously compounded interest rate. Use 0.01 for 1% :type r: float :param sigma: Annualized volatility. Use 0.01 for 1% :type sigma: float :param q: Annualized dividend yield. Use 0.01 for 1% :type q: float

Return type:

float

pyqstrat.pyqstrat_cpp.delta(call: numpy.ndarray[bool], S: numpy.ndarray[numpy.float64], K: numpy.ndarray[numpy.float64], t: numpy.ndarray[numpy.float64], r: numpy.ndarray[numpy.float64], sigma: numpy.ndarray[numpy.float64], q: numpy.ndarray[numpy.float64] = 0.0) → object

Compute European option delta :param call: True for a call option, False for a put :type call: bool :param S: Spot price. For a future discount the future price using exp(-rt) :type S: float :param K: Strike :type K: float :param t: Time to maturity in years :type t: float :param r: Continuously compounded interest rate. Use 0.01 for 1% :type r: float :param sigma: Annualized volatility. Use 0.01 for 1% :type sigma: float :param q: Annualized dividend yield. Use 0.01 for 1%. Default 0 :type q: float, optional

Returns:

Option delta

Return type:

float

pyqstrat.pyqstrat_cpp.gamma(S: numpy.ndarray[numpy.float64], K: numpy.ndarray[numpy.float64], t: numpy.ndarray[numpy.float64], r: numpy.ndarray[numpy.float64], sigma: numpy.ndarray[numpy.float64], q: numpy.ndarray[numpy.float64] = 0.0) → object

Compute European option gamma. :param S: Spot price. For a future discount the future price using exp(-rt) :type S: float :param K: Strike :type K: float :param t: Time to maturity in years :type t: float :param r: Continuously compounded interest rate. Use 0.01 for 1% :type r: float :param sigma: Annualized volatility. Use 0.01 for 1% :type sigma: float :param q: Annualized dividend yield. Use 0.01 for 1%. Default 0 :type q: float, optional

Returns:

Option gamma

Return type:

float

pyqstrat.pyqstrat_cpp.implied_vol(call: numpy.ndarray[bool], price: numpy.ndarray[numpy.float64], S: numpy.ndarray[numpy.float64], K: numpy.ndarray[numpy.float64], t: numpy.ndarray[numpy.float64], r: numpy.ndarray[numpy.float64], q: numpy.ndarray[numpy.float64] = 0.0) → object

Compute implied volatility for a European option. :param call: True for a call option, False for a put :type call: bool :param price: The option premium :type price: float :param S: Spot price. For a future discount the future price using exp(-rt) :type S: float :param K: Strike :type K: float :param t: Time to maturity in years :type t: float :param r: Continuously compounded interest rate. Use 0.01 for 1% :type r: float :param q: Annualized dividend yield. Use 0.01 for 1%. Default 0 :type q: float, optional

Returns:

Implied volatility. For 1% we return 0.01

Return type:

float

pyqstrat.pyqstrat_cpp.rho(call: numpy.ndarray[bool], S: numpy.ndarray[numpy.float64], K: numpy.ndarray[numpy.float64], t: numpy.ndarray[numpy.float64], r: numpy.ndarray[numpy.float64], sigma: numpy.ndarray[numpy.float64], q: numpy.ndarray[numpy.float64] = 0.0) → object

Compute European option rho. This is Black Scholes formula rho divided by 100 so we get rho per 1% change in interest rate :param call: True for a European call option, False for a put :type call: bool :param S: Spot price. For a future discount the future price using exp(-rt) :type S: float :param K: Strike :type K: float :param t: Time to maturity in years :type t: float :param r: Continuously compounded interest rate. Use 0.01 for 1% :type r: float :param sigma: Annualized volatility. Use 0.01 for 1% :type sigma: float :param q: Annualized dividend yield. Use 0.01 for 1%. Default 0 :type q: float, optional

Returns:

Option theta

Return type:

float

pyqstrat.pyqstrat_cpp.theta(call: numpy.ndarray[bool], F: numpy.ndarray[numpy.float64], K: numpy.ndarray[numpy.float64], t: numpy.ndarray[numpy.float64], r: numpy.ndarray[numpy.float64], sigma: numpy.ndarray[numpy.float64], q: numpy.ndarray[numpy.float64] = 0.0) → object

Compute European option theta per day. This is Black Scholes formula theta divided by 365 to give us the customary theta per day :param call: True for a call option, False for a put :type call: bool :param S: Spot price. For a future discount the future price using exp(-rt) :type S: float :param K: Strike :type K: float :param t: Time to maturity in years :type t: float :param r: Continuously compounded interest rate. Use 0.01 for 1% :type r: float :param sigma: Annualized volatility. Use 0.01 for 1% :type sigma: float :param q: Annualized dividend yield. Use 0.01 for 1%. Default 0 :type q: float, optional

Returns:

Option theta

Return type:

float

pyqstrat.pyqstrat_cpp.vega(S: float, K: float, t: float, r: float, sigma: float, q: float = 0.0) → float

Compute European option vega. This is Black Scholes formula vega divided by 100 so we get rho per 1% change in interest rate :param S: Spot price. For a future discount the future price using exp(-rt) :type S: float :param K: Strike :type K: float :param t: Time to maturity in years :type t: float :param r: Continuously compounded interest rate. Use 0.01 for 1% :type r: float :param sigma: Annualized volatility. Use 0.01 for 1% :type sigma: float :param q: Annualized dividend yield. Use 0.01 for 1%. Default 0 :type q: float, optional

Returns:

Option vega

Return type:

float

pyqstrat.pyqstrat_io module

pyqstrat.pyqstrat_io.parse_datetimes()

parse datetimes

pyqstrat.pyqstrat_io.read_file()

read a file

pyqstrat.markets module

class pyqstrat.markets.EminiFuture

Bases: object

calendar = <pyqstrat.holiday_calendars.Calendar object>

static get_current_symbol(curr_date)

>>> assert(EminiFuture.get_current_symbol(datetime.date(2019, 3, 14)) == 'ESH9')
:rtype: :py:class:`str`

>>> assert(EminiFuture.get_current_symbol(datetime.date(2019, 3, 15)) == 'ESM9')
>>> assert(EminiFuture.get_current_symbol(datetime.date(2020, 3, 14)) == 'ESH0')

static get_expiry(fut_symbol)

>>> assert(EminiFuture.get_expiry('ESH8') == np.datetime64('2018-03-16T08:30'))
:rtype: :py:class:`~numpy.datetime64`

static get_next_symbol(curr_future_symbol)

>>> assert(EminiFuture.get_next_symbol('ESZ8') == 'ESH9')
:rtype: :py:class:`str`

static get_previous_symbol(curr_future_symbol)

>>> assert(EminiFuture.get_previous_symbol('ESH9') == 'ESZ8')
:rtype: :py:class:`str`

class pyqstrat.markets.EminiOption

Bases: object

calendar = <pyqstrat.holiday_calendars.Calendar object>

static decode_symbol(name)

Return type:

tuple[weekday, int, int, int]

>>> EminiOption.decode_symbol('E1AF8')
(MO, 2018, 1, 1)

static get_expiry(symbol)

>>> EminiOption.get_expiry('EW2Z5')
numpy.datetime64('2015-12-11T15:00')
>>> EminiOption.get_expiry('E3AF7')
numpy.datetime64('2017-01-17T15:00')
:rtype: :py:class:`~numpy.datetime64`

>>> EminiOption.get_expiry('EWF0')
numpy.datetime64('2020-01-31T15:00')

pyqstrat.markets.future_code_to_month(future_code)

Given a future code such as “X”, return the month abbreviation, such as “nov”

Parameters:

future_code (str) – the one letter future code

Return type:

str

>>> future_code_to_month('X')
'nov'

pyqstrat.markets.future_code_to_month_number(future_code)

Given a future code such as “X”, return the month number (from 1 - 12)

Parameters:

future_code (str) – the one letter future code

Return type:

int

>>> future_code_to_month_number('X')
11

pyqstrat.markets.get_future_code(month)

Given a month number such as 3 for March, return the future code for it, e.g. H >>> get_future_code(3) ‘H’

Return type:

str

pyqstrat.strategy_builder module

class pyqstrat.strategy_builder.StrategyBuilder(data=None)

Bases: object

A helper class that makes it easier to build simpler strategies by removing the need for a lot of boilerplate code. The __call__ member function returns a Strategy built by this class that can then be run and evaluated.

Parameters:

• data (Optional[DataFrame]) – A pandas dataframe containing columns for timestamps, indicators and signals

• timestamps – The ‘’heartbeat’’ of the strategy. A vector of each relevant time for this strategy

• timestamp_unit – Corresponds to bar length. Follows numpy datetime64 datatype strings. For example if you have daily bars, use M8[D] and if you have minute bars use M8[m]

• contract_groups – Each contract has to be a part of a contract group for PNL aggregation, etc. This class builds a contract group for each contract that is supplied to it, so this is only needed if you want to group contracts

• price_function – A function of type PriceFunctionType that we use to get market prices by contract name and timestamp

• pnl_calc_time – Since pnl is reported on a daily basis, this indicates the time (in minutes) for computing that PNL. For example, if you want to compute PNL at 4:01 pm each day, use 16 * 60 + 1

• starting_equity – Equity to start the backtest with

• trade_lag – How long it takes for an order to get to the market (in bars). If you are using input data from 2:33:44 pm it may take you less than one second to generate the order and send it to the market. In this case, assuming you have 1 second bars, set the trade lag to 1. Market simulator get orders this many bars after the order is generated by trading rules. Trade lag of 0 can be used for daily orders, when you want to relax the assumption that it takes finite time to generate an order and send it to the market

• strategy_context – Properties that are passed into all user defined functions such as rule functions, signal functions. For example, you may want to set a property that indicates commission per trade instead of hardcoding it into a market simulator.

• indicators – Indicators are numeric vectors used to generate trading signals. For example, we may create an indicator that has the overnight gap (difference between yesterday’s closing price and this morning’s open price)

• signals – Signals are boolean vectors used to determine whether or not we should run rules. For example, we may want to run an entry rule if the indicator above is more than 1% of last night’s closing price. In this case the signal value for the corresponding time bar would be set to True

• rules – Rules are functions that take indicators and signals as inputs and generate orders as outputs at each time bar. Depending on the logic in each rule, it decides sizing, what kind of order to generate (limit, market, etc.) and how many (or zero) orders to generate.

• market_sims – Market simulators simulate the execution of orders and generation of trades. The market sims are called in order so two market simulators should not successfully process the same order

__call__()

Generates a strategy object that we can then run and evaluate

Return type:

Strategy

__init__(data=None)

add_contract(symbol)

Return type:

Contract

add_contract_group(contract_group)

Return type:

None

add_indicator(name, indicator, contract_groups=None, depends_on=None)

Return type:

None

add_market_sim(market_sim_function)

Return type:

None

add_rule(name, rule_function, signal_name, sig_true_values=None, position_filter=None)

Return type:

None

add_series_indicator(name, column_name)

Return type:

None

add_series_rule(column_name, rule_function, position_filter='', name='', contract_groups=None)

Helper function that makes it easier to create a signal by using a boolean column from the dataframe passed in the constructor to this class

Return type:

None

add_signal(name, signal_function, contract_groups=None, depends_on_indicators=None, depends_on_signals=None)

Return type:

None

contract_groups: dict[str, ContractGroup]

data: DataFrame

indicators: list[tuple[str, Callable[[ContractGroup, ndarray, SimpleNamespace, SimpleNamespace], ndarray], Optional[Sequence[ContractGroup]], Optional[Sequence[str]]]]

log_orders: bool

log_trades: bool

market_sims: list[Callable[[Sequence[Order], int, ndarray, dict[str, SimpleNamespace], dict[str, SimpleNamespace], SimpleNamespace], list[Trade]]]

pnl_calc_time: int

price_function: Optional[Callable[[Contract, ndarray, int, SimpleNamespace], float]]

rules: list[tuple[str, Callable[[ContractGroup, int, ndarray, SimpleNamespace, ndarray, Account, Sequence[Order], SimpleNamespace], list[Order]], str, Optional[Sequence[Any]], Optional[str]]]

set_log_orders(log_orders)

Return type:

None

set_log_trades(log_trades)

Return type:

None

set_pnl_calc_time(pnl_calc_time)

Return type:

None

set_price_function(price_function)

Return type:

None

set_starting_equity(starting_equity)

Return type:

None

set_strategy_context(context)

Return type:

None

set_timestamps(timestamps)

Return type:

None

set_trade_lag(trade_lag)

Return type:

None

signals: list[tuple[str, Callable[[ContractGroup, ndarray, SimpleNamespace, SimpleNamespace, SimpleNamespace], ndarray], Optional[Sequence[ContractGroup]], Optional[Sequence[str]], Optional[Sequence[str]]]]

starting_equity: float

strategy_context: SimpleNamespace

timestamp_unit: dtype

timestamps: Optional[ndarray]

trade_lag: int

pyqstrat.strategy_components module

class pyqstrat.strategy_components.BracketOrderEntryRule(reason_code, price_func, long=True, percent_of_equity=0.1, min_stop_return=0, max_position_size=0, single_entry_per_day=False, contract_filter=None, stop_return_func=None)

Bases: object

A rule that generates orders with stops

Parameters:

• reason_code (str) – Reason for the orders created used for display

• price_func (Callable[[Contract, ndarray, int, SimpleNamespace], float]) – A function that returns price given a contract and timestamp

• long (bool) – whether we want to go long or short

• percent_of_equity (float) – How much to risk per trade as a percentage of current equity. Used to calculate order qty so that if we get stopped out, we don’t lose more than this amount. Of course if price gaps up or down rather than moving smoothly, we may lose more.

• stop_return_func (Optional[Callable[[Contract, ndarray, int, SimpleNamespace], float]]) – A function that gives us the distance between entry price and stop price

• min_stop_return (float) – We will not enter if the stop_return is closer than this percent to the stop. Otherwise, we get very large trade sizes

• max_position_size (float) – An order should not result in a position that is greater than this percent of the portfolio

• contract_filter (Optional[Callable[[ContractGroup, int, ndarray, SimpleNamespace, ndarray, Account, Sequence[Order], SimpleNamespace], list[str]]]) – A function that takes similar arguments as a rule (with ContractGroup) replaced by Contract but returns a list of contract names for each positive signal timestamp. For example, for a strategy that trades 5000 stocks, you may want to construct a single signal and apply it to different contracts at different times, rather than create 5000 signals that will call your rule 5000 times every time the signal is true.

>>> timestamps = np.arange(np.datetime64('2023-01-01'), np.datetime64('2023-01-05'))
>>> sig_values = np.full(len(timestamps), False)
>>> aapl_prices = np.array([100.1, 100.2, 100.3, 100.4])
>>> ibm_prices = np.array([200.1, 200.2, 200.3, 200.4])
>>> aapl_stops = np.array([-0.5, -0.3, -0.2, -0.01])
>>> ibm_stops=  np.array([-0.5, -0.3, -0.2, -0.15])
>>> price_dict = {'AAPL': (timestamps, aapl_prices), 'IBM': (timestamps, ibm_prices)}
>>> stop_dict = {'AAPL': (timestamps, aapl_stops), 'IBM': (timestamps, ibm_stops)}
>>> price_func = PriceFuncArrayDict(price_dict)
>>> fr = BracketOrderEntryRule('TEST_ENTRY', price_func, long=False)
>>> default_cg = ContractGroup.get('DEFAULT')
>>> default_cg.clear_cache()
>>> default_cg.add_contract(Contract.get_or_create('AAPL'))
>>> default_cg.add_contract(Contract.get_or_create('IBM'))
>>> account = SimpleNamespace()
>>> account.equity = lambda x: 1e6
>>> orders = fr(default_cg, 1, timestamps, SimpleNamespace(), sig_values, account, [], SimpleNamespace())
>>> assert len(orders) == 2 and orders[0].qty == -998 and orders[1].qty == -499
>>> stop_return_func = PriceFuncArrayDict(stop_dict)
>>> fr = BracketOrderEntryRule('TEST_ENTRY', price_func, long=True, stop_return_func=stop_return_func, min_stop_return=-0.1)
>>> orders = fr(default_cg, 2, timestamps, SimpleNamespace(), sig_values, account, [], SimpleNamespace())
>>> assert len(orders) == 2 and orders[0].qty == 4985 and orders[1].qty == 2496
>>> orders = fr(default_cg, 3, timestamps, SimpleNamespace(), sig_values, account, [], SimpleNamespace())
>>> assert len(orders) == 1 and orders[0].qty == 3326

__call__(contract_group, i, timestamps, indicator_values, signal_values, account, current_orders, strategy_context)

Call self as a function.

Return type:

list[Order]

__init__(reason_code, price_func, long=True, percent_of_equity=0.1, min_stop_return=0, max_position_size=0, single_entry_per_day=False, contract_filter=None, stop_return_func=None)

contract_filter: Optional[Callable[[ContractGroup, int, ndarray, SimpleNamespace, ndarray, Account, Sequence[Order], SimpleNamespace], list[str]]]

long: bool

max_position_size: float

min_stop_returnt: float

percent_of_equity: float

price_func: Callable[[Contract, ndarray, int, SimpleNamespace], float]

reason_code: str

single_entry_per_day: bool

stop_return_func: Optional[Callable[[Contract, ndarray, int, SimpleNamespace], float]]

class pyqstrat.strategy_components.ClosePositionExitRule(reason_code, price_func, limit_increment=nan)

Bases: object

A rule to close out an open position

Parameters:

• reason_code (str) – the reason for closing out used for display purposes

• price_func (Callable[[Contract, ndarray, int, SimpleNamespace], float]) – the function this rule uses to get market prices

• limit_increment (float) – if not nan, we add or subtract this number from current market price (if selling or buying respectively) and create a limit order

__call__(contract_group, i, timestamps, indicator_values, signal_values, account, current_orders, strategy_context)

Call self as a function.

Return type:

list[Order]

__init__(reason_code, price_func, limit_increment=nan)

limit_increment: float

price_func: Callable[[Contract, ndarray, int, SimpleNamespace], float]

reason_code: str

class pyqstrat.strategy_components.PercentOfEquityTradingRule(reason_code, price_func, equity_percent=0.1, long=True, allocate_risk=False, limit_increment=nan)

Bases: object

A rule that trades a percentage of equity.

Parameters:

• reason_code (str) – Reason for entering the order, used for display

• equity_percent (float) – Percentage of equity used to size order

• long (bool) – Whether We want to go long or short

• allocate_risk (bool) – If set, we divide the max risk by number of trades. Otherwise each trade will be alloated max risk

• limit_increment (float) – If not nan, we add or subtract this number from current market price (if selling or buying respectively) and create a limit order. If nan, we create market orders

• price_func (Callable[[Contract, ndarray, int, SimpleNamespace], float]) – The function we use to get intraday prices

__call__(contract_group, i, timestamps, indicator_values, signal_values, account, current_orders, strategy_context)

Call self as a function.

Return type:

list[Order]

__init__(reason_code, price_func, equity_percent=0.1, long=True, allocate_risk=False, limit_increment=nan)

allocate_risk: bool = False

equity_percent: float = 0.1

limit_increment: float = nan

long: bool = True

price_func: Callable[[Contract, ndarray, int, SimpleNamespace], float]

reason_code: str

class pyqstrat.strategy_components.PriceFuncArrayDict(price_dict, allow_previous=False)

Bases: object

A function object with a signature of PriceFunctionType and takes a dictionary of

contract name -> tuple of sorted timestamps and prices

Parameters:

• price_dict (dict[str, tuple[ndarray, ndarray]]) – a dict with key=contract nane and value a tuple of timestamp and price arrays

• allow_previous (bool) – if set and we don’t find an exact match for the timestamp, use the previous timestamp. Useful if you have a dict with keys containing dates instead of timestamps

>>> timestamps = np.arange(np.datetime64('2023-01-01'), np.datetime64('2023-01-04'))
>>> price_dict = {'AAPL': (timestamps, [8, 9, 10]), 'IBM': (timestamps, [20, 21, 22])}
>>> pricefunc = PriceFuncArrayDict(price_dict)
>>> Contract.clear_cache()
>>> aapl = Contract.create('AAPL')
>>> assert(pricefunc(aapl, timestamps, 2, None) == 10)
>>> ibm = Contract.create('IBM')
>>> basket = Contract.create('AAPL_IBM', components=[(aapl, 1), (ibm, -1)])
>>> assert(pricefunc(basket, timestamps, 1, None) == -12)

__call__(contract, timestamps, i, context)

Call self as a function.

Return type:

float

__init__(price_dict, allow_previous=False)

allow_previous: bool

price_dict: dict[str, tuple[ndarray, ndarray]]

class pyqstrat.strategy_components.PriceFuncArrays(symbols, timestamps, prices, allow_previous=False)

Bases: object

A function object with a signature of PriceFunctionType. Takes three ndarrays of symbols, timestamps and prices

__call__(contract, timestamps, i, context)

Call self as a function.

Return type:

float

__init__(symbols, timestamps, prices, allow_previous=False)

allow_previous: bool

price_dict: dict[str, tuple[ndarray, ndarray]]

class pyqstrat.strategy_components.PriceFuncDict(price_dict)

Bases: object

A function object with a signature of PriceFunctionType and takes a dictionary of contract name -> timestamp -> price >>> timestamps = np.arange(np.datetime64(‘2023-01-01’), np.datetime64(‘2023-01-04’)) >>> aapl_prices = [8, 9, 10] >>> ibm_prices = [20, 21, 22] >>> price_dict = {‘AAPL’: {}, ‘IBM’: {}} >>> for i, timestamp in enumerate(timestamps): … price_dict[‘AAPL’][timestamp] = aapl_prices[i] … price_dict[‘IBM’][timestamp] = ibm_prices[i] >>> pricefunc = PriceFuncDict(price_dict) >>> Contract.clear_cache() >>> aapl = Contract.create(‘AAPL’) >>> assert(pricefunc(aapl, timestamps, 2, None) == 10) >>> ibm = Contract.create(‘IBM’) >>> basket = Contract.create(‘AAPL_IBM’, components=[(aapl, 1), (ibm, -1)]) >>> assert(pricefunc(basket, timestamps, 1, None) == -12)

__call__(contract, timestamps, i, context)

Call self as a function.

Return type:

float

__init__(price_dict)

price_dict: dict[str, dict[datetime64, float]]

class pyqstrat.strategy_components.SimpleMarketSimulator(price_func, slippage_pct=0.0, commission=0, price_rounding=3, post_trade_func=None)

Bases: object

A function object with a signature of MarketSimulatorType. It can take into account slippage and commission >>> ContractGroup.clear_cache() >>> Contract.clear_cache() >>> put_symbol, call_symbol = ‘SPX-P-3500-2023-01-19’, ‘SPX-C-4000-2023-01-19’ >>> put_contract = Contract.create(put_symbol) >>> call_contract = Contract.create(call_symbol) >>> basket = Contract.create(‘test_contract’) >>> basket.components = [(put_contract, -1), (call_contract, 1)] >>> timestamp = np.datetime64(‘2023-01-03 14:35’) >>> price_func = PriceFuncDict({put_symbol: {timestamp: 4.8}, call_symbol: {timestamp: 3.5}}) >>> order = MarketOrder(contract=basket, timestamp=timestamp, qty=10, reason_code=’TEST’) >>> sim = SimpleMarketSimulator(price_func=price_func, slippage_pct=0) >>> out = sim([order], 0, np.array([timestamp]), {}, {}, SimpleNamespace()) >>> assert(len(out) == 1) >>> assert(math.isclose(out[0].price, -1.3)) >>> assert(out[0].qty == 10)

__call__(orders, i, timestamps, indicators, signals, strategy_context)

TODO: code for stop orders

Return type:

list[Trade]

__init__(price_func, slippage_pct=0.0, commission=0, price_rounding=3, post_trade_func=None)

Parameters:

• price_func (Callable[[Contract, ndarray, int, SimpleNamespace], float]) – A function that we use to get the price to execute at

• slippage_pct (float) – Slippage per dollar transacted. Meant to simulate the difference between bid/ask mid and execution price

• commission (float) – Fee paid to broker per trade

commission: float

post_trade_func: Optional[Callable[[Trade, SimpleNamespace], None]]

price_func: Callable[[Contract, ndarray, int, SimpleNamespace], float]

price_rounding: int

slippage_pct: float

class pyqstrat.strategy_components.StopReturnExitRule(reason_code, price_func, stop_return_func)

Bases: object

A rule that exits any given positions if a stop is hit. You should set entry_price in the strategy context in the market simulator when you enter a position

__call__(contract_group, i, timestamps, indicator_values, signal_values, account, current_orders, context)

Call self as a function.

Return type:

list[Order]

__init__(reason_code, price_func, stop_return_func)

price_func: Callable[[Contract, ndarray, int, SimpleNamespace], float]

reason_code: str

stop_return_func: Callable[[Contract, ndarray, int, SimpleNamespace], float]

class pyqstrat.strategy_components.VWAPCloseRule(vwap_minutes, reason_code)

Bases: object

Rule to close out a position at vwap price :type reason_code: str :param reason_code: Reason_code: Reason for each order. For display purposes :type vwap_minutes: int :param vwap_minutes: How long the vwap period is. For example, a 5 minute vwap order will execute at 5 minute vwap

__call__(contract_group, i, timestamps, indicator_values, signal_values, account, current_orders, strategy_context)

Call self as a function.

Return type:

list[Order]

__init__(vwap_minutes, reason_code)

reason_code: str

vwap_minutes: int

class pyqstrat.strategy_components.VWAPEntryRule(reason_code, vwap_minutes, price_func, long=True, percent_of_equity=0.1, stop_price_ind=None, min_price_diff_pct=0, single_entry_per_day=False)

Bases: object

A rule that generates VWAP orders

Parameters:

• reason_code (str) – Reason for each order. For display purposes

• vwap_minutes (int) – How long the vwap period is. For example, a 5 minute vwap order will execute at 5 minute vwap

• market (from when it is sent to the) –

• price_func (Callable[[Contract, ndarray, int, SimpleNamespace], float]) – A function that this rule uses to get market price at a given timestamp

• long (bool) – Whether we want to go long or short

• percent_of_equity (float) – Order qty is calculated so that if the stop price is reached, we lose this amount

• stop_price_ind (Optional[str]) – Don’t enter if estimated entry price is market price <= stop price + min_price_diff_pct * market_price (for long orders) or the opposite for short orders

• min_price_diff_pct (float) – See stop_price_ind

__call__(contract_group, i, timestamps, indicator_values, signal_values, account, current_orders, strategy_context)

Call self as a function.

Return type:

list[Order]

__init__(reason_code, vwap_minutes, price_func, long=True, percent_of_equity=0.1, stop_price_ind=None, min_price_diff_pct=0, single_entry_per_day=False)

long: bool

min_price_diff_pct: float

percent_of_equity: float

price_func: Callable[[Contract, ndarray, int, SimpleNamespace], float]

reason_code: str

single_entry_per_day: bool

stop_price_ind: Optional[str]

vwap_minutes: int

class pyqstrat.strategy_components.VWAPMarketSimulator(price_indicator, volume_indicator, backup_price_indicator=None)

Bases: object

A market simulator that simulates buying and selling at VWAP prices. This works with VWAP orders and ignores all other order types The order executes either: a. After the vwap end time defined in the VWAP order b. If marker price <= vwap stop price defined in the VWAP order for buy orders c. If market price >= vwap stop price for sell orders

__call__(orders, i, timestamps, indicators, signals, strategy_context)

Call self as a function.

Return type:

list[Trade]

__init__(price_indicator, volume_indicator, backup_price_indicator=None)

Parameters:

• price_indicator (str) – An indicator that contains historical trade price per timestamp

• volume_indicator (str) – An indicator that contains volume per timestamp

• backup_price_indicator (Optional[str]) – Execution price to use if price or volume is missing

backup_price_indicator: Optional[str]

price_indicator: str

volume_indicator: str

class pyqstrat.strategy_components.VectorIndicator(vector)

Bases: object

An indicator created from a vector :type vector: ndarray :param vector: Vector with indicator values. Must be the same length as strategy timestamps

__call__(contract_group, timestamps, indicator_values, context)

Call self as a function.

Return type:

ndarray

__init__(vector)

vector: ndarray

class pyqstrat.strategy_components.VectorSignal(vector)

Bases: object

A signal created from a vector that has boolean values :type vector: ndarray :param vector: Vector with indicator values. Must be the same length as strategy timestamps

__call__(contract_group, timestamps, indicator_values, parent_values, context)

Call self as a function.

Return type:

ndarray

__init__(vector)

vector: ndarray

pyqstrat.strategy_components.get_contract_price_from_array_dict(price_dict, contract, timestamp, allow_previous)

Return type:

float

pyqstrat.strategy_components.get_contract_price_from_dict(price_dict, contract, timestamp)

Return type:

float

pyqstrat.interactive_plot module

class pyqstrat.interactive_plot.InteractivePlot(data, labels=None, transform_func=<pyqstrat.interactive_plot.SimpleTransform object>, create_selection_widgets_func=<function create_selection_dropdowns>, dim_filter_func=<function simple_dimension_filter>, data_filter_func=<function simple_data_filter>, stat_func=<pyqstrat.interactive_plot.MeanWithCI object>, plot_func=<pyqstrat.interactive_plot.LineGraphWithDetailDisplay object>, display_form_func=<function display_form>, debug=False)

Bases: object

Creates a multidimensional interactive plot off a dataframe.

__init__(data, labels=None, transform_func=<pyqstrat.interactive_plot.SimpleTransform object>, create_selection_widgets_func=<function create_selection_dropdowns>, dim_filter_func=<function simple_dimension_filter>, data_filter_func=<function simple_data_filter>, stat_func=<pyqstrat.interactive_plot.MeanWithCI object>, plot_func=<pyqstrat.interactive_plot.LineGraphWithDetailDisplay object>, display_form_func=<function display_form>, debug=False)

Parameters:

• data (DataFrame) – The pandas dataframe to use for plotting

• labels (Optional[dict[str, str]]) – A dict where column names from the dataframe are mapped to user friendly labels. For any column names not found as keys in this dict, we use the column name as the label. Default None

• dim_filter_func (Callable[[DataFrame, str, list[tuple[str, Any]]], ndarray]) – A function that generates the values of a dimension based on other dimensions. For example, if the user chooses “Put Option” in a put/call dropdown, the valid strikes could change in a Strike dropdown that follows. Default simple_dimension_filter

• data_filter_func (Callable[[DataFrame, list[tuple[str, Any]]], DataFrame]) – A function that filters the data to plot. For example, if the user chooses “Put Option” in a put/call dropdown, we could filter the dataframe to only include put options. Default simple_data_filter

• stat_func (Callable[[DataFrame, str, str, str], list[tuple[str, DataFrame, dict[Any, DataFrame]]]]) – Once we have filtered the data, we may need to plot some statistics, such as mean and confidence intervals. In this function, we compute these statistics. Default MeanWithCI()

• plot_func (Callable[[str, str, list[tuple[str, DataFrame, dict[Any, DataFrame]]]], list[Widget]]) – A function that plots the data. This could also display detail data used to compute the statistics associated with each data point.

• display_form_func (Callable[[Sequence[Widget], bool], None]) – A function that displays the form given a list of plotly widgets (including the graph widget)

• debug – Dont clear forms if this is true so we can see print output

create_pivot(xcol, ycol, zcol, dimensions)

Create the initial pivot :type xcol: str :param xcol: Column name to use as the x axis in the DataFrame :type ycol: str :param ycol: Column name to use as the y axis in the DataFrame :type zcol: str :param zcol: Column name to use for z-values. Each zvalue can be used for a different trace within this plot. For example, a column :param called “option_type” could contain the values “American”: :param “European”: :param “Bermudan” and we could plot the data for each type: :param in a separate trace: :type dimensions: dict[str, Any] :param dimensions: The column names used for filter dimensions. For example, we may want to filter by days to expiration and put/call :param The key the column name and the value is the initial value for that column. For example: :param in a: :param dropdown for Put/Call we may want “Put” to be the initial value set in the dropdown. Set to None if you: :param don’t care what initial value is chosen.:

Return type:

None

update(owner_idx=-1)

Redraw the form using the values of all widgets above and including the one with index owner_idx. If owner_idx is -1, we redraw everything.

Return type:

None

class pyqstrat.interactive_plot.LineConfig(color=None, thickness=nan, secondary_y=False, marker_mode='lines+markers', show_detail=True)

Bases: object

__init__(color=None, thickness=nan, secondary_y=False, marker_mode='lines+markers', show_detail=True)

color: Optional[str] = None

marker_mode: str = 'lines+markers'

secondary_y: bool = False

show_detail: bool = True

thickness: float = nan

class pyqstrat.interactive_plot.LineGraphWithDetailDisplay(display_detail_func=<pyqstrat.interactive_plot.SimpleDetailTable object>, line_configs={}, title=None, hovertemplate=None, debug=False)

Bases: object

Draws line graphs and also includes a detail pane. When you click on a point on the line graph, the detail pane shows the data used to compute that point.

__call__(xaxis_title, yaxis_title, line_data)

Draw the plot and also set it up so if you click on a point, we display the data used to compute that point. :type line_data: list[tuple[str, DataFrame, dict[Any, DataFrame]]] :param line_data: The zvalue, plot data, and detail data for each line to draw. The plot data must have :param x as the first column and y as the second column: :rtype: list[Widget]

Return:

A list of widgets to draw. In this case, a figure widget and a output widget which contains the detail display

__init__(display_detail_func=<pyqstrat.interactive_plot.SimpleDetailTable object>, line_configs={}, title=None, hovertemplate=None, debug=False)

Parameters:

• display_detail_func (Callable[[Widget, DataFrame, bool], None]) – A function that displays the data on the detail pane. Default SimpleDetailTable

• line_configs (dict[str, LineConfig]) – Configuration of each line. The key in this dict is the zvalue for that line. Default {}

• title (Optional[str]) – Title of the graph. Default None

• hovertemplate (Optional[str]) – What to display when we hover over a point on the graph. See plotly hovertemplate

class pyqstrat.interactive_plot.MeanWithCI(mean_func=<function nanmean>, ci_level=0)

Bases: object

Computes mean (or median) and optionally confidence intervals for plotting

__call__(filtered_data, xcol, ycol, zcol)

For each unique value of x and z, compute mean (and optionally ci) of y. :rtype: list[tuple[str, DataFrame, dict[Any, DataFrame]]] :returns: x, y data for plotting lines of the mean of y versus x for each z and the data used to compute the mean

__init__(mean_func=<function nanmean>, ci_level=0)

Parameters:

• mean – The function to compute ci for

• ci_level (int) – Set to 0 for no confidence intervals, or the level you want. For example, set to 95 to compute a 95% confidence interval. Default 0

class pyqstrat.interactive_plot.SimpleDetailTable(colnames=None, float_format='{:.4g}', min_rows=100, copy_to_clipboard=True)

Bases: object

Displays a pandas DataFrame under a plot that contains the data used to compute a statistic of y for each x, y pair

__call__(detail_widget, data, debug=False)

Parameters:

• detail_widget (Widget) – The widget to display the data in

• data (DataFrame) – The dataframe to display

Return type:

None

__init__(colnames=None, float_format='{:.4g}', min_rows=100, copy_to_clipboard=True)

Parameters:

• colnames (Optional[list[str]]) – list of column names to display. If None we display all columns. Default None

• float_format (str) – Format for each floating point column. Default {:.4g}

• min_rows (int) – Do not truncate the display of the table before this many rows. Default 100

• copy_to_clipboard (bool) – If set, we copy the dataframe to the clipboard. On linux, you must install xclip for this to work

class pyqstrat.interactive_plot.SimpleTransform(transforms=None)

Bases: object

Initial transformation of data. For example, you might add columns that are quantiles of other columns

__call__(data)

Call self as a function.

Return type:

DataFrame

__init__(transforms=None)

class pyqstrat.interactive_plot.TestInteractivePlot(methodName='runTest')

Bases: TestCase

test_interactive_plot()

transform(data)

Return type:

DataFrame

pyqstrat.interactive_plot.create_selection_dropdowns(dims, labels, update_form_func)

Create a list of selection widgets

Return type:

dict[str, Any]

pyqstrat.interactive_plot.display_form(form_widgets, debug=False)

Return type:

None

pyqstrat.interactive_plot.foo(name, old, new)

pyqstrat.interactive_plot.on_widgets_updated(change, update_form_func, selection_widgets)

Callback called by plotly when widgets are updated by the user.

Return type:

None

pyqstrat.interactive_plot.percentile_buckets(a, n=10)

>>> np.random.seed(0)
:rtype: :py:class:`~numpy.ndarray`

>>> a = np.random.uniform(size=10000)
>>> assert np.allclose(np.unique(percentile_buckets(a)), np.arange(0.05, 1, 0.1), atol=0.01)

pyqstrat.interactive_plot.simple_data_filter(data, selected_values)

Filters a dataframe based on the selected values

Return type:

DataFrame

pyqstrat.interactive_plot.simple_dimension_filter(data, dim_name, selected_values)

Produces a list to put into a dropdown for selecting a dimension value

Return type:

ndarray

Module contents