HumanRupert

# src/ms/controller.py

tickers: List[Constituent] = convert_csv_to_records(
        "data/tickers.csv", Constituent)

dt_to_milli = lambda dt: datetime.timestamp(dt) * 1000
start = dt_to_milli(datetime(2018, 1, 1))
end = dt_to_milli(datetime(2020, 1, 1))

extract_n_store_cup_with_handles(start, end, tickers)

HumanRupert

Start date	2016-01-04
End date	2017-12-29
Total months	23

Backtest
---------
Annual return	9.7%
Cumulative returns	20.2%
Annual volatility	7.5%
Sharpe ratio	1.27

HumanRupert

WATCHLIST_WINDOW_DAYS = 30
ABOVE_PIVOT_PCT = 1.01
TAKE_PROFIT_PCT = 1.15
STOP_LOSS_PCT = .95
PATIENCE_WINDOW_DAYS = 21
START = datetime(2016, 1, 1)
END = datetime(2018, 1, 1)
BENCHMARK = "SPY"
SHORT_MA_LEN = 50
LONG_MA_LEN = 200

HumanRupert

# format start end
to_localized_ts = lambda dt: pd.Timestamp(dt).tz_localize("UTC")
start, end = to_localized_ts(START), to_localized_ts(END)

# get returns
benchmark = get_benchmark_returns()

# run strat
results = zp.run_algorithm(
    start=start,

HumanRupert

def analyze(perf: pd.DataFrame, bench: pd.Series) -> None:
    returns, positions, transactions = pf.utils.extract_rets_pos_txn_from_zipline(perf)
    pf.create_full_tear_sheet(returns=returns, benchmark_rets=bench)

HumanRupert

def get_benchmark_returns() -> pd.Series:
    bench = yf.Ticker(BENCHMARK)
    bench_hist = bench_hist.history(start=START, end=END, auto_adjust=True).tz_localize("UTC")
    returns = pd.Series(bench_hist["Close"].pct_change().values, index=bench_hist.index).dropna()
    returns.index.names = ["date"]
    return returns

HumanRupert

def handle_data(context, data):
    current_dt = zp.api.get_datetime()

    prices = data.history(context.stocks, "price", bar_count=200, frequency="1d")
    
    # look for new trades
    for ix, pattern in context.patterns.iterrows():
        # skip if asset is already in portfolio
        open_positions = set(context.portfolio.positions.keys())
        symbol = zp.api.symbol(pattern["symbol"])

HumanRupert

def initialize(context):
    # avoid out of bounds error by dropping firstBottomDate col
    patterns = pd.read_csv("data/patterns.csv").drop(["firstBottomDate"], axis=1)
    patterns = convert_date_cols(patterns)
    context.patterns = patterns

    tickers = pd.read_csv("data/tickers.csv")
    tickers = convert_date_cols(tickers)
    context.stocks = [zp.api.symbol(ticker) for ticker in tickers.symbol]

HumanRupert

def convert_date_cols(df: pd.DataFrame) -> pd.DataFrame:
    """Given a dataframe, adds UTC timezone to all columns that have date in their names."""
    for col in df.columns:
        if("date" in col.lower()):
            df[col] = pd.to_datetime(df[col]).dt.tz_localize("UTC")
    return df

HumanRupert

WATCHLIST_WINDOW_DAYS = 30
ABOVE_PIVOT_PCT = 1.01
TAKE_PROFIT_PCT = 1.15
STOP_LOSS_PCT = .95
PATIENCE_WINDOW_DAYS = 21
START = datetime(2016, 1, 1)
END = datetime(2018, 1, 1)
BENCHMARK = "SPY"
SHORT_MA_LEN = 50
LONG_MA_LEN = 200