Files
2022-02-19 17:13:55 -08:00

657 lines
26 KiB
Python

# -*- coding: utf-8 -*-
import datetime as dt
from random import choice as rChoice
from numpy import absolute, any, concatenate, cumsum, flip, max
from numpy import mean, min, ndarray, std, sum, where, zeros
from numpy.random import choice, normal, randint
from pandas import DataFrame, date_range
from pandas_ta._typing import Array, Float, Int, IntFloat, List, Optional
from pandas_ta.maps import Imports, RATE
class sample(object):
"""Sample Data [sample] BETA (Only core features so far)
DISCLAIMER: Use at your own risk!
This is a Numpy and stochastics package wrapper Class that easily creates
stochastic process realization with or without stochastic noise.
To get the most out of sample(), install the 'stochastic' package:
$ pip install stochastic
The following stochastic package noise and processes have been
implemented:
* Noise[9]: Blue "b", Brownian "br, Fractal Gaussian "fg", Gaussian "g",
Pink "p", Red "r", Violet "v", Wiener "w", Random "rand", or None
* Processes[11]: Brownian Bridge "bb", Brownian Excursion "be",
Brownian Meander "bm", Brownian Motion "bmo",
Cox Ingersoll Ross "cir", Fractional Brownian Motion "fbm",
Geometric Brownian Motion "gbm", Ornstein Uhlenbeck "ou",
Random Walk "rw", Wiener "w", Random "rand" or None.
* If the stochastic process is not installed, a Simple Random Walk is
realized without noise.
* When argument process="rand", a process is chosen at random.
* When argument noise="rand", a noise is chosen at random.
Sources:
https://stochastic.readthedocs.io/en/stable/
Args:
* Basic options
name (str): Set a ticker name. Default: A random ticker like 'SMPL'.
process (str): The process to realize. See options above.
Default: None
noise (str): Noise to apply. See options above. Default: None
length (int): How many observations to generate.
Default: ta.RATE["TRADING_DAYS_PER_YEAR"] (252)
* Additional transformations
orient (str): Applies either a Reversal, Inversion or
Inverted Reversal of the realization. Default: None.
positive (bool): If the resultant process is non-negative.
Default: True
scale (str): Applies either Mean, Normal or Standard scale.
Default: None.
noise_percent (float): Percentage of noise to apply. (Not implemented)
Default: 1.0
* Arguments specific to the stochastics package
s0 (float): The initial value of the process. Default: 0.01
b (float): The "b" argument for Brownian Bridge and Meander.
Default: 0.01
t (float): The "t" arguments of certain processes. Default: 0.01
drift (float): The drift for some processes. For Cox-Ingersoll-Ross,
mean = drift. Default: 0
volatility (float): The volatility for some processes. For Brownian
Motion, scale = volatility. Default: 1
speed (float): The speed value for some processes. Default: 1
hurst (float): The Hurst value for Fractional Brownian Motion "fbm".
Default: 0.5
steps (list): A list of step increments for the Random Walk.
Default: [-1, 1]
random_number (int): Random number for the stochastic package to use.
Default: None
* Misc. Options
future (bool): Whether the resultant DataFrame Index has a future
date range or a past date range. Default: True
freq (str): The frequency to use for the generated DataFrame
date range index. (Not implemented) In Default: "D"
intraday (str): If intraday is 'full', 24 hours, or is an 'equity'
with 6.5 hours. (Not implemented) Default: "full"
date_fmt (str): Date Format for Daterange. Default: '%Y-%m-%d'
precision (int): How many decimals to round when printing to stdout.
Default: 6
verbose (bool): Show more info to stdout. Default: False
Examples:
(A) Returns a Brownian Motion Process class instance with initial value
(s0) 12.34, drift 0.2, scale (volatility) 0.4, t of 5, and random seed of 21
>>> sp = ta.sample(name="SMPL", process="bmo", s0=12.34, drift=0.2, volatility=0.4, t=5, random_number=21)
Returns Numpy values of the process
>>> sp.np
Returns a DataFrame of the process
>>> sp.df
Reverse the process (returns numpy array)
>>> rev_sp = sp.orientation(sp.np, "r")
Standard Rescaling of the process (returns numpy array)
>>> scaled_sp = sp.scale(sp.np, "s")
Some class properties
Returns list of processes available
>>> sp.processes
Returns selected process
>>> sp.process
Returns list of noises available
>>> sp.noises
Returns selected noise
>>> sp.process
(B) Returns a Random Process and Noise with initial value (s0) 1, positive
values, inverted, mean scaled and verbose
>>> rpn = ta.sample(s0=1.0, process="rand", noise="rand", positive=True, orient="i", scale="m", verbose=True)
>>> rpn.np # numpy values
>>> rpn.df # Pandas DataFrame
"""
_noises = ["b", "br", "fg", "g", "p", "r", "v", "w", None, "rand"]
_orientations = ["i", "r", "ir", "ri", None, "rand"]
_processes = [
"bb", "be", "bm", "bmo", "cir", "fbm",
"gbm", "ou", "rw", "w", None, "rand"
]
_scales = ["m", "n", "s", None]
def __init__(self,
name: str = None, process: str = None, noise: str = None,
length: Int = None, s0: IntFloat = None, b: IntFloat = None,
t: IntFloat = None, drift: Int = None, volatility: IntFloat = None,
speed: IntFloat = None, hurst: Float = None,
steps: List[IntFloat] = None, random_number: Optional[Int] = None,
orient: str = None, positive: bool = None, scale: str = None,
future: bool = None, freq: str = None, intraday: str = None,
noise_percent: Float = None, date_fmt: str = None,
precision: Int = None, verbose: bool = None
):
"""Validation and initialization of arguments and then runs the
_generate() method to build a sample realization with the given
arguments.
"""
_random_symbol = ''.join(
[rChoice("ABCDEFGHIJKLMNOPQRSTUVWXYZ") for _ in range(randint(3, 6))])
self._name = str(name) if name is not None and isinstance(
name, str) else _random_symbol
self._process = str(process).lower() if process is not None and isinstance(
process, str) and process in self._processes else None
self._noise = str(noise).lower() if noise is not None and isinstance(
noise, str) and noise in self._noises else None
self._length = int(length) if isinstance(
length, int) else RATE["TRADING_DAYS_PER_YEAR"]
self._s0 = float(s0) if s0 is not None and isinstance(
s0, (float, int)) else 0.01
self._b = float(b) if b is not None and isinstance(b, float) else 0.0
self._t = float(t) if t is not None and isinstance(
t, (float, int)) else 1.0
self._drift = float(drift) if drift is not None and isinstance(
drift, (float, int)) else 0.0
self._volatility = float(volatility) if volatility is not None and isinstance(
volatility, (float, int)) else 1.0
self._speed = float(speed) if speed is not None and isinstance(
speed, (float, int)) else 1.0
self._hurst = float(hurst) if hurst is not None and isinstance(
hurst, float) and 0 <= hurst <= 1 else 0.5
self._steps = steps if steps is not None and isinstance(
steps, list) and len(steps) > 1 else [-1.0, 1.0]
self._random_number = random_number if random_number is not None else None
self._orient = str(orient).lower() if orient is not None and isinstance(
orient, str) and orient in self._orientations else None
self._positive = positive if positive is not None and isinstance(
positive, bool) else False
self._scale = str(scale).lower() if scale is not None and isinstance(
scale, str) and scale in self._scales else None
self._future = future if future is not None and isinstance(
future, bool) else False
self._freq = f"{freq.lower()}" if freq is not None and len(
freq) else "D"
self._intraday = intraday.lower() if intraday is not None and len(
intraday) and intraday in ["full", "equity"] else "full"
self._noise_percent = float(noise_percent) if noise_percent is not None and isinstance(
noise_percent, float) else 1.0 # Percent as decimal
_date_fmt = str(noise) if date_fmt is not None and isinstance(
date_fmt, str) else "%Y-%m-%d"
self._date_today = dt.date.today().strftime(_date_fmt)
self._precision = int(precision) if precision is not None and isinstance(
precision, int) else 6
self._verbose = verbose if verbose is not None and isinstance(
verbose, bool) else False
if self._process == "rand":
self._process = choice(self._processes[:-2])
if self._noise == "rand":
self._noise = choice(self._noises[:-1])
self._generate() # Run it
def _bernoulli_mask(self,
array: Array, percent: Float = None, p: Float = None
):
"""Bernoulli Mask - Positive or Negative"""
if array.size > 0:
percent = float(percent) if percent is not None and isinstance(
percent, float) else self.noise_percent
p = float(p) if p is not None and isinstance(
p, float) and p > 0 and p < 1 else 0.5
return array * self.noise_percent * self._bernoulli_process()
return array
def _bernoulli_process(self):
"""Bernoulli Process"""
return randint(2, size=self.length)
def _generate(self):
"""A method to generate stochastic process realizations.
Order of operations:
1. Generate a stochastic process and noise if not None and combine.
2. Reorient original realization is not None.
3. Make realization non-negative if an values are <= 0 if the
'positive' argument is True.
4. Recale the realization if the not None.
5. Save the result to self._np.
"""
values = self._stoch_process()
values += self._stoch_noise()
if self._orient is not None and isinstance(self._orient, str):
values = self._orientation(values, mode=self._orient)
if self._verbose:
print(f"[i] Orientation: {self._orient}")
if self.positive:
_s0 = values[0]
values = self._nonnegative(values)
if self._verbose:
print(
f"[i] New s0: {round(values[0], self._precision)} from {round(_s0, self._precision)} (change {round(values[0] - _s0, self._precision)})")
if self._scale is not None and isinstance(self._scale, str):
values = self._scaler(values, self._scale)
if self._verbose:
print(f"[i] Scaled to: {self._scale}")
self._np = values
_npns = f"{self.name} | {self.process} {self.noise+' ' if self.noise is not None else ''}{self.np.size}"
_s0n = f"s0: {round(self.np[0], self._precision)}, sN: {round(self.np[-1], self._precision)}"
_msmm = f"mu: {round(mean(self.np), self._precision)}, sigma: {round(std(self.np), self._precision)}"
self._dfname = f"{_npns} | {_s0n} | {_msmm}"
if self._verbose:
print(self._dfname)
def nonnegative(self, array: Array = None):
"""Vertical Translation the 'array' where the resultant 'array' has
non-negative values."""
if isinstance(array, ndarray):
return self._nonnegative(array)
return array
def _nonnegative(self, array: Array):
"""Translates the array up by the minimum of the 'array' if any values
are negative."""
if array.size > 0 and any(array < 0):
array += -1 * array.min()
self._s0 = array[0]
return array
def _normal_mask(self, array: Array):
"""A method to add some additional randomness to the realized
process. Applies a mask based on the Normal Distribution and the 'array's
mean and standard deviation."""
if array.size > 0:
norm = normal(mean(array), std(array), size=self.length)
return array * self.noise_percent * norm
return array
def orientation(self, array: Array, mode: str = None):
"""Orients the 'array' either by Inversion, Reversal, or an
Inverted Reversal."""
if isinstance(array, ndarray):
return self._orientation(array, mode=mode)
return array
def _orientation(self, array: Array, mode: str = None):
"""Orients the 'array' either by Inversion, Reversal, or an
Inverted Reversal."""
_modes = ["i", "r", "ir", "ri", None, "rand"]
mode = str(mode).lower() if mode is not None and isinstance(
mode, str) and mode.lower() in _modes else None
result = array
if mode is None:
return result
if mode == "rand":
mode = choice(_modes[3:])
if mode == "i":
mid = 0.5 * (min(array) + max(array))
inv = mid - array
diff = inv - inv[0]
result = array[0] + diff if array[0] > 0 else diff - array[0]
if mode == "r":
result = flip(array) - (array[-1] - array[0])
if mode in ["ir", "ri"]:
result = self._orientation(self._orientation(array, "i"), "r")
return result
def scale(self, array: Array, mode: str):
"""Mean, Normal or Standard scaling of the 'array'."""
if isinstance(array, ndarray):
return self._scaler(array, mode=mode)
return array
def _scaler(self, array: Array, mode: str):
"""Scaling: mean, normal, standard"""
result = array
if mode is None:
return result
if mode == "rand":
mode = choice(self._scales[3:])
min_, max_ = min(array), max(array)
range_ = absolute(max_ - min_)
mu_, std_ = mean(array), std(array)
if mode == "m" and range_ > 0: # "mean"
result = ((array - mu_) / range_)
if mode == "n" and range_ > 0: # "normal"
result = ((array - min_) / range_)
if mode == "s" and std_ > 0: # "standard"
result = ((array - mu_) / std_)
return result
def _simple_random_walk(self,
up: Float = None, down: Float = None
) -> Array:
"""Simple Random Walk
Sources:
https://sphelps.net/teaching/scf/slides/random-walks-slides.html
"""
up = float(up) if up is not None and isinstance(
up, (int, float)) else 1.0
down = float(down) if down is not None and isinstance(
down, (int, float)) else -1.0
if up < down:
down, up = up, down
x = concatenate(([0.0],
where(randint(0, 2, size=self.length - 1) == 0, down, up)
))
return cumsum(x).astype(float)
def _stoch_noise(self):
"""Method to apply noise from the stochastic package if installed.
Otherwise, it returns 0 noise.
"""
_desc = f"[+] "
result = zeros(self.length, dtype=float)
if self._noise is not None and Imports["stochastic"]:
from stochastic import random as st_random
st_random.use_generator()
st_random.seed(self.random_number)
if self._noise in ["blue", "b"]:
from stochastic.processes.noise import BlueNoise
result = BlueNoise(t=self.t).sample(self.length - 1)
_desc += f"Blue Noise [blue|b] | t: {self.t}"
elif self._noise in ["brownian", "br"]:
from stochastic.processes.noise import BrownianNoise
result = BrownianNoise(t=self.t).sample(self.length - 1)
_desc += f"Brownian Noise [brownian|br] | t: {self.t}"
elif self._noise in ["fractional", "fg"]:
from stochastic.processes.noise.fractional_gaussian_noise import FractionalGaussianNoise
result = FractionalGaussianNoise(
hurst=self.hurst, t=self.t).sample(
self.length)
_desc += f"Fractional Gaussian Noise [fractal|fg] | hurst: {self.hurst}, t: {self.t}"
elif self._noise in ["gauss", "g"]:
from stochastic.processes.noise import GaussianNoise
result = GaussianNoise(t=self.t).sample(self.length)
_desc += f"Gaussian Noise [gauss|g] | t: {self.t}"
elif self._noise in ["pink", "p"]:
from stochastic.processes.noise import PinkNoise
result = PinkNoise(t=self.t).sample(self.length - 1)
_desc += f"Pink Noise [pink|p] | t: {self.t}"
elif self._noise in ["red", "r"]:
from stochastic.processes.noise import RedNoise
result = RedNoise(t=self.t).sample(self.length - 1)
_desc += f"Red Noise [red|r] | t: {self.t}"
elif self._noise in ["violet", "v"]:
from stochastic.processes.noise import VioletNoise
result = VioletNoise(t=self.t).sample(self.length - 1)
_desc += f"Violet Noise [violet|v] | t: {self.t}"
elif self._noise in ["white", "w"]:
from stochastic.processes.noise import WhiteNoise
result = WhiteNoise(t=self.t).sample(self.length - 1)
_desc += f"White Noise [white|w] | t: {self.t}"
else:
_desc = "Noiseless"
else: # Default if no stochastic package installed
_desc = "srw"
# Initial Value (s0) adjustment
result = result + \
result[0] if result[0] > self.s0 else result - result[0]
if result is not None and any(result) and self._verbose:
print(_desc)
return result
def _stoch_process(self):
"""Method to return some realizations from the stochastic package.
Otherwise, it returns a Simple Random Walk."""
_desc = f"[+] "
result = None
if self._process is not None and Imports["stochastic"]:
from stochastic import random as st_random
st_random.use_generator()
st_random.seed(self.random_number)
if self._process == "bb":
from stochastic.processes.continuous import BrownianBridge
result = self.s0 + \
BrownianBridge(b=self.b, t=self.t).sample(self.length - 1)
_desc += f"Brownian Bridge [bb] | s0: {self.s0}, b: {self.b}, t: {self.t}"
elif self._process == "be":
from stochastic.processes.continuous import BrownianExcursion
result = self.s0 + \
BrownianExcursion(t=self.t).sample(self.length - 1)
_desc += f"Brownian Excursion [be] | s0: {self.s0}, t: {self.t}"
elif self._process == "bm":
from stochastic.processes.continuous import BrownianMeander
result = self.s0 + \
BrownianMeander(t=self.t).sample(self.length - 1, self.b)
_desc += f"Brownian Meander [bm] | s0: {self.s0}, t: {self.t}"
elif self._process == "bmo":
from stochastic.processes.continuous import BrownianMotion
result = self.s0 + BrownianMotion(
drift=self.drift,
scale=self.volatility,
t=self.t).sample(
self.length - 1)
_desc += f"Brownian Motion [bmo] | s0: {self.s0}, drift: {self.drift}, scale: {self.volatility}, t: {self.t}"
elif self._process == "cir":
from stochastic.processes.diffusion import CoxIngersollRossProcess
result = CoxIngersollRossProcess(
speed=self.speed,
mean=self.drift,
vol=self.volatility,
t=self.t).sample(
self.length - 1,
self.s0)
_desc += f"Cox Ingersoll Ross [cir] | s0: {self.s0}, speed: {self.speed}, mean: {self.drift}, vol: {self.volatility}, t: {self.t}"
elif self._process == "fbm":
from stochastic.processes.continuous import FractionalBrownianMotion
result = self.s0 + \
FractionalBrownianMotion(
hurst=self.hurst, t=self.t).sample(
self.length - 1)
_desc += f"Fractal Brownian Motion [fbm] | s0: {self.s0}, hurst: {self.hurst}, t: {self.t}"
elif self._process == "gbm":
from stochastic.processes.continuous import GeometricBrownianMotion
result = GeometricBrownianMotion(
drift=self.drift,
volatility=self.volatility,
t=self.t).sample(
self.length - 1,
self.s0)
_desc += f"Geometric Brownian Motion [gbm] | s0: {self.s0}, drift: {self.drift}, vol: {self.volatility}, t: {self.t}"
elif self._process == "ou":
from stochastic.processes.diffusion import OrnsteinUhlenbeckProcess
result = OrnsteinUhlenbeckProcess(
speed=self.speed, vol=self.volatility, t=self.t).sample(
self.length - 1, self.s0)
_desc += f"Ornstein Uhlenbeck [ou] | s0: {self.s0}, speed: {self.speed}, vol: {self.volatility}, t: {self.t}"
elif self._process == "rw":
from stochastic.processes.discrete import RandomWalk
result = self.s0 + \
RandomWalk(
steps=self.steps).sample(
self.length -
1).astype(float)
_desc += f"Random Walk [rw] | s0: {self.s0}, steps: {self.steps}"
elif self._process == "w":
from stochastic.processes.continuous import WienerProcess
result = self.s0 + \
WienerProcess(t=self.t).sample(self.length - 1)
_desc += f"Wiener [w] | s0: {self.s0}, t: {self.t}"
else: # Default if no stochastic package installed
result = self.s0 + self._simple_random_walk()
_desc += f"Simple Random Walk | s0: {self.s0}"
if result is not None and self._verbose:
print(_desc)
return result
@property
def b(self):
"""The 'b' value for some stochastic processes."""
return self._b
@property
def datetime_range(self):
"""The Pandas datetimerange used by the resultant DataFrame index."""
if hasattr(self, "_datetimerange") and self._datetimerange is not None:
return self._datetimerange
else:
self._datetimerange = None
# datelist = date_range(date_N_days_ago, periods=N, freq=freq).to_pydatetime().tolist()
if self.future:
self._datetimerange = date_range(
start=self._date_today, periods=self.length, freq=self.freq)
else:
self._datetimerange = date_range(
end=self._date_today, periods=self.length, freq=self.freq)
return self._datetimerange
@property
def df(self):
"""The Pandas DataFrame of the sample/realization."""
if self.np is not None and self.np.size > 0:
df = DataFrame(
self.np,
index=self.datetime_range,
columns=["close"])
df.name = self._dfname
self._df = df
return self._df
return None
@property
def drift(self):
"""The 'drift' value for some stochastic processes."""
return self._drift
@property
def freq(self):
"""The frequency of the Pandas daterange."""
return self._freq
@property
def future(self):
"""Whether the Pandas daterange DataFrame index has future values or
past values from today."""
return self._future
@property
def hurst(self):
"""The 'Hurst' value for Fractional Brownian Motion 'fbm'."""
return self._hurst
@property
def length(self):
"""The length of the sample/realization."""
return self._length
@property
def name(self):
"""The name of the sample/realization.
If not set, randomly generated with prefix '~'.
"""
return self._name
@property
def noise(self):
"""Noise sample/realization."""
return self._noise
@property
def noise_percent(self):
"""Percent to apply to noise."""
return self._noise_percent
@property
def noises(self):
"""List of available noises in the class."""
return self._noises
@property
def np(self):
"""The Numpy values of the sample/realization."""
return self._np
@property
def orient(self):
"""The orientation applied when generated."""
return self._orient
@property
def positive(self):
"""Enforce non-negative values of the sample/realization."""
return self._positive
@property
def process(self):
"""Process sample/realization."""
return self._process
@property
def processes(self):
"""List of available processes in the class."""
return self._processes
@property
def random_number(self):
"""Random Number of the sample/realization."""
return self._random_number
@property
def s0(self):
"""The initial value of the stochastic process."""
return self._s0
@property
def speed(self):
"""The 'speed' value for some stochastic processes."""
return self._speed
@property
def steps(self):
"""The 'steps' value for Random Walk."""
return self._steps
@property
def t(self):
"""The 't' value for some stochastic processes."""
return self._t
@property
def volatility(self):
"""The 'volatility' value for some stochastic processes."""
return self._volatility