Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def run_backtest(self):
self.log.notice("Running in backtest mode")
try:
run_algorithm(
algo_namespace=self.id,
capital_base=self.trading_info["CAPITAL_BASE"],
data_frequency=self.trading_info["DATA_FREQ"],
initialize=self._init_func,
handle_data=self._process_data,
analyze=self._analyze,
exchange_name=self.trading_info["EXCHANGE"],
quote_currency=self.trading_info["QUOTE_CURRENCY"],
start=pd.to_datetime(self.trading_info["START"], utc=True),
end=pd.to_datetime(self.trading_info["END"], utc=True),
)
except exchange_errors.PricingDataNotLoadedError as e:
self.log.critical("Failed to run stratey Requires data ingestion")
raise e
# from kryptos.worker import ingester
def run_backtest(self):
self.log.notice("Running in backtest mode")
try:
run_algorithm(
algo_namespace=self.id,
capital_base=self.trading_info["CAPITAL_BASE"],
data_frequency=self.trading_info["DATA_FREQ"],
initialize=self._init_func,
handle_data=self._process_data,
analyze=self._analyze,
exchange_name=self.trading_info["EXCHANGE"],
quote_currency=self.trading_info["BASE_CURRENCY"],
start=pd.to_datetime(self.trading_info["START"], utc=True),
end=pd.to_datetime(self.trading_info["END"], utc=True),
)
except exchange_errors.PricingDataNotLoadedError as e:
self.log.critical("Failed to run stratey Requires data ingestion")
raise e
# from kryptos.worker import ingester
self.log.warning(
f"End Date: {end_arrow} is invalid, will use 30 minutes from now"
)
end_arrow = arrow.utcnow().shift(minutes=+5)
self.trading_info["END"] = end_arrow.datetime
self.log.debug(
f"Stopping strategy {end_arrow.humanize()} -- {end_arrow.datetime}"
)
# catalyst loads state before init called
# so need to fetch state before algorithm starts
if outputs.load_state_from_storage(self):
self.log.warning(f"Resuming strategy with saved state")
run_algorithm(
capital_base=self.trading_info["CAPITAL_BASE"],
initialize=self._init_func,
handle_data=self._process_data,
analyze=self._analyze,
exchange_name=self.trading_info["EXCHANGE"],
live=True,
algo_namespace=self.id,
quote_currency=self.trading_info["QUOTE_CURRENCY"],
live_graph=False,
simulate_orders=simulate_orders,
stats_output=None,
# start=pd.to_datetime(start.datetime, utc=True),
end=pd.to_datetime(end_arrow.datetime, utc=True),
auth_aliases=auth_aliases,
)
getLast(analysis, 'stoch_over_bought')))
log.info(
'- stoch_over_sold: {}'.format(getLast(analysis, 'stoch_over_sold')))
log.info('- rsi_over_bought: {}'.format(
getLast(analysis, 'rsi_over_bought')))
log.info(
'- rsi_over_sold: {}'.format(getLast(analysis, 'rsi_over_sold')))
def getLast(arr, name):
return arr[name][arr[name].index[-1]]
if __name__ == '__main__':
run_algorithm(
capital_base=10000,
data_frequency='daily',
initialize=initialize,
handle_data=handle_data,
analyze=analyze,
exchange_name='poloniex',
quote_currency='usdt',
start=pd.to_datetime('2016-11-1', utc=True),
end=pd.to_datetime('2017-11-10', utc=True),
)
prices = data.current(context.assets, 'price')
record(price=prices)
pass
def analyze(context, perf):
stats = get_pretty_stats(perf)
print(stats)
pass
if __name__ == '__main__':
live = True
if live:
run_algorithm(
capital_base=0.01,
initialize=initialize,
handle_data=handle_data,
exchange_name='poloniex',
algo_namespace='buy_btc_polo_jh',
quote_currency='btc',
analyze=analyze,
live=True,
simulate_orders=True,
)
else:
run_algorithm(
capital_base=1000,
data_frequency='daily',
initialize=initialize,
handle_data=handle_data,
# Replace all NA, NAN or infinite values with its nearest value
def fill(series):
if isinstance(series, pd.Series):
return series.replace([np.inf, -np.inf], np.nan).ffill().bfill()
elif isinstance(series, np.ndarray):
return pd.Series(series).replace([np.inf, -np.inf],
np.nan).ffill().bfill().values
else:
return series
if __name__ == '__main__':
start_date = pd.to_datetime('2017-01-08', utc=True)
end_date = pd.to_datetime('2017-11-13', utc=True)
performance = run_algorithm(start=start_date, end=end_date,
capital_base=10000.0,
initialize=initialize,
handle_data=handle_data,
analyze=analyze,
exchange_name='poloniex',
data_frequency='minute',
base_currency='btc',
live=False,
live_graph=False,
algo_namespace='simple_universe')
"""
Run in Terminal (inside catalyst environment):
def run_algo(initialize, handle_data, analyze):
try:
run_algorithm(
capital_base=CONFIG["CAPITAL_BASE"],
data_frequency=CONFIG["DATA_FREQ"],
initialize=initialize,
handle_data=handle_data,
analyze=analyze,
exchange_name=CONFIG["EXCHANGE"],
quote_currency=CONFIG["BASE_CURRENCY"],
start=pd.to_datetime(CONFIG["START"], utc=True),
end=pd.to_datetime(CONFIG["END"], utc=True),
)
except PricingDataNotLoadedError:
log.info("Ingesting required exchange bundle data")
load.ingest_exchange(CONFIG)
plt.legend(loc=3)
start, end = ax6.get_ylim()
ax6.yaxis.set_ticks(np.arange(0, end, end / 5))
# Show the plot.
plt.gcf().set_size_inches(18, 8)
plt.show()
pass
if __name__ == '__main__':
# The execution mode: backtest or live
live = False
if live:
run_algorithm(
capital_base=0.025,
initialize=initialize,
handle_data=handle_data,
analyze=analyze,
exchange_name='poloniex',
live=True,
algo_namespace=NAMESPACE,
quote_currency='btc',
live_graph=False,
simulate_orders=False,
stats_output=None,
)
else:
folder = os.path.join(
tempfile.gettempdir(), 'catalyst', NAMESPACE
# catalyst run -f catalyst/examples/mean_reversion_simple.py -x poloniex -s 2017-7-1 -e 2017-7-31 -c usdt -n mean-reversion --data-frequency minute --capital-base 10000
run_algorithm(
capital_base=1,
data_frequency='minute',
initialize=initialize,
handle_data=handle_data,
analyze=analyze,
exchange_name='bitfinex',
algo_namespace=algo_namespace,
base_currency='usd',
start=pd.to_datetime('2017-10-1', utc=True),
end=pd.to_datetime('2017-11-13', utc=True),
)
elif MODE == 'live':
run_algorithm(
initialize=initialize,
handle_data=handle_data,
analyze=analyze,
exchange_name='bitfinex',
live=True,
algo_namespace=algo_namespace,
base_currency='usd',
live_graph=True
)