What do I have to subclass to make zipline pull data from my personal data api instead of a bundle?

[Brett Elliot]

Hi,

I have personal server with daily and minute data and my own API to access it. What do I have to subclass, extend, modify, etc to get zipline to use my data api instead of bundles?

Thanks!

[hamed shafiee]

Hi,

Did you find any solution for this please?

Thanks

[Kanat Mergenbayev]

from zipline.api import \
    schedule_function, \
    get_open_orders, \
    order_target_percent, \
    order, \
    get_datetime, \
    symbol, \
    date_rules, \
    time_rules, \
    record, \
    set_slippage

from pandas import read_sql_query, Timedelta
from zipline.finance import slippage
from sqlalchemy import create_engine
from collections import OrderedDict
from datetime import datetime

import pandas as pd
import sqlalchemy as sqla
import jsonpickle
import zipline
import sys


ticker = sys.argv[1]
db_params = sys.argv[2]
strategy = sys.argv[3]


class BackTestBuy(object):

    def __init__(self, params):
        self.ticker = params["ticker"]

        self.start_session = params["start_session"]
        self.end_session = params["end_session"]
        self.test_data = params["minute_test_data"]
        self.five_minute_test_data = params["five_minute_test_data"]

        self.performance = self.get_performance()

    def set_initial_context_params(self, context):

        "Set context params here"

    def initialize(self, context):

        context.set_slippage(slippage.NoSlippage())
        context.scheduled_data = {}
        self.set_initial_context_params(context)

        schedule_function(self.my_record_vars,
                          date_rules.every_day(),
                          time_rules.market_open())

    def my_record_vars(self, context, data):

        context.stop_trades = context.limit_trades = 1
        self.set_initial_context_params(context)

        if self.position_exists(context) and self.order_not_exists():
            order_target_percent(symbol(self.ticker), 0)

    def position_not_exists(self, context):

        if context.portfolio.positions[symbol(self.ticker)].amount == 0:
            return True
        else:
            return False

    def position_exists(self, context):

        if context.portfolio.positions[symbol(self.ticker)].amount != 0:
            return True
        else:
            return False

    def position(self, context):

        return context.portfolio.positions[symbol(self.ticker)].amount

    def trade_price(self, context):

        return context.portfolio.positions[symbol(self.ticker)].cost_basis

    def order_not_exists(self):

        if not get_open_orders(symbol(self.ticker)):
            return True
        else:
            return False

    def order_exists(self):

        if get_open_orders(symbol(self.ticker)):
            return True
        else:
            return False

    def get_last_fifth_minute(self, current_time):

        if current_time.minute % 5 == 0:
            return (current_time - Timedelta(minutes=5)).tz_convert("UTC")
        else:
            return (current_time - Timedelta(minutes=(current_time.minute % 5))).tz_convert("UTC")

    def five_minute_condition(self, current_time, current_price):

        df = self.five_minute_test_data

        if df.empty:
            return False
        else:
            df = df.loc[df['date'] == self.get_last_fifth_minute(current_time)]
            if df.empty:
                return False
            else:
                print(" return True or False on condition or data process further")

    def handle_data(self, context, data):

        current_price = data.current(symbol(self.ticker), 'price')

        current_time = pd.Timestamp(get_datetime()).tz_convert('US/Eastern')

        if self.order_not_exists() and self.position_not_exists(context)\
                and self.five_minute_condition(current_time, current_price):
            order_target_percent(symbol(self.ticker), 1)

        if self.order_not_exists() and self.position_exists()\
                and self.five_minute_condition(current_time, current_price):
            order_target_percent(symbol(self.ticker), 0)

    def get_performance(self):

        return zipline.run_algorithm(start=self.start_session,
                                     end=self.end_session,
                                     initialize=self.initialize,
                                     capital_base=100000,
                                     handle_data=self.handle_data,
                                     data_frequency='minute',
                                     data=self.test_data)


def error_function(connection, ticker, strategy, test_param, e):

    e = str(e).replace("'", "")
    connection.execute("INSERT INTO back_test_error(date, ticker, strategy, test_param, error) "
                       "VALUES('" + str(datetime.now()) + "', '" +
                       ticker + "', '" +
                       strategy + "', '" +
                       test_param + "', '"
                       + e + "')")


def refactor_orders(xs):

    result = []

    for item in xs:

        item["dt"] = item["dt"].strftime("%Y-%m-%d %H:%M:%S")
        item["created"] = item["created"].strftime("%Y-%m-%d %H:%M:%S")

        result.append(item)

    return result


def refactor_transactions(xs):

    result = []

    for item in xs:

        item["dt"] = item["dt"].strftime("%Y-%m-%d %H:%M:%S")
        item["price"] = item["price"].item()

        result.append(item)

    return result


def write_result_to_database(connection, ticker, strategy, df, time_frame):

    try:
        connection.execute("DELETE FROM back_test_result "
                           " WHERE ticker = '" + ticker + "'"
                           " AND strategy = '" + strategy + "'")
    except Exception as e:
        error_function(connection, "delete back test result", e)
    metadata = sqla.MetaData(bind=connection)
    datatable = sqla.Table("back_test_result", metadata, autoload=True)
    for index, item in df.iterrows():
        stmt = datatable.insert().values(date=index, ticker=ticker, time_frame=time_frame,
                                         strategy=strategy,
                                         algo_volatility=item["algo_volatility"],
                                         algorithm_period_return=item["algorithm_period_return"],
                                         alpha=item["alpha"],
                                         benchmark_period_return=item["benchmark_period_return"],
                                         benchmark_volatility=item["benchmark_volatility"],
                                         beta=item["beta"],
                                         capital_used=item["capital_used"],
                                         ending_cash=item["ending_cash"],
                                         ending_exposure=item["ending_exposure"],
                                         ending_value=item["ending_value"],
                                         excess_return=item["excess_return"],
                                         gross_leverage=item["gross_leverage"],
                                         long_exposure=item["long_exposure"],
                                         long_value=item["long_value"],
                                         longs_count=item["longs_count"],
                                         max_drawdown=item["max_drawdown"],
                                         max_leverage=item["max_leverage"],
                                         net_leverage=item["net_leverage"],
                                         orders=jsonpickle.encode(refactor_orders(item["orders"]))
                                         if item["orders"] else None,
                                         period_close=item["period_close"],
                                         period_label=item["period_label"],
                                         period_open=item["period_open"],
                                         pnl=item["pnl"],
                                         portfolio_value=item["portfolio_value"],
                                         positions=jsonpickle.encode(item["positions"]) if item["positions"] else None,
                                         returns=item["returns"],
                                         sharpe=item["sharpe"],
                                         short_exposure=item["short_exposure"],
                                         short_value=item["short_value"],
                                         shorts_count=item["shorts_count"],
                                         sortino=item["sortino"],
                                         starting_cash=item["starting_cash"],
                                         starting_exposure=item["starting_exposure"],
                                         starting_value=item["starting_value"],
                                         trading_days=item["trading_days"],
                                         transactions=jsonpickle.encode(refactor_transactions(item["transactions"]))
                                         if item["transactions"] else None,
                                         treasury_period_return=item["treasury_period_return"])
        try:
            connection.execute(stmt)
        except Exception as e:
            error_function(connection, "write back test result", e)


def get_minute_test_data(connection, ticker):

    data = OrderedDict()

    data[ticker] = read_sql_query("SELECT * FROM test_minute_data "
                                  "WHERE ticker = '" + ticker + "' "
                                  "ORDER BY date ASC",
                                  connection,
                                  parse_dates=["date"],
                                  index_col="date")

    data[ticker] = data[ticker][["open", "high", "low", "close", "volume"]]

    panel = pd.Panel(data)
    panel.minor_axis = ["open", "high", "low", "close", "volume"]
    panel.major_axis = panel.major_axis.tz_convert("UTC")
    panel.major_axis = panel.major_axis.tz_convert(None)

    return panel


def get_five_minute_test_data(connection, ticker):

    return read_sql_query("SELECT * "
                          "FROM test_five_minute_data "
                          "WHERE ticker = '" + ticker + "' "
                          "ORDER BY date DESC",
                          connection,
                          parse_dates=["date"])


def start_date(connection, ticker, table):

    return read_sql_query("SELECT date FROM " + table + " "
                          "WHERE ticker = '" + ticker + "' "
                          "ORDER BY date ASC LIMIT 1",
                          connection).values


def end_date(connection, ticker, table):

    return read_sql_query("SELECT date FROM " + table + " "
                          "WHERE ticker = '" + ticker + "' "
                          "ORDER BY date DESC LIMIT 1",
                          connection).values


def initial_test_params(connection, ticker,):

    return get_minute_test_data(connection, ticker),\
           (start_date(connection, ticker, "test_minute_data") + Timedelta(days=1)).to_pydatetime()[0], \
           end_date(connection, ticker, "test_minute_data").to_pydatetime()[0], \
           get_five_minute_test_data(connection, ticker)


engine = create_engine(db_params)
connection = engine.connect()

write_result_to_database(connection,
                         ticker,
                         strategy,
                         BackTestBuy(initial_test_params(connection, ticker)).performance,
                         "minute")

connection.close()
engine.dispose()