Skip to content

Instantly share code, notes, and snippets.

@robcarver17

robcarver17/functions.py Secret

Created May 11, 2021
Embed
What would you like to do?
import pandas as pd
from ib_insync import Future, IB, Forex
from syscore.objects import missing_data
## MODIFY SO RETURNS *ALL* RELEVANT STUFF!
## DO A FULL RUN WITH *ALL* INSTRUMENTS INCLUDING MISSING SGX
## DOUBLE CHECK ALL FIGURES
def identify_duplicates(ib, symbol, exchange):
print("%s %s" % (symbol, exchange))
future = Future(symbol=symbol, exchange = exchange)
contracts = ib.reqContractDetails(future)
if len(contracts)==0:
print("Missing data for %s/%s" % (symbol, exchange))
return missing_data
list_of_ccy= [contract.contract.currency for contract in contracts]
list_of_multipliers = [contract.contract.multiplier for contract in contracts]
unique_list_of_ccy = list(set(list_of_ccy))
unique_list_of_multipliers= list(set(list_of_multipliers))
if len(unique_list_of_multipliers)>1:
print("%s/%s more than one multiplier %s" %
(symbol, exchange, str(unique_list_of_multipliers)))
multiplier = str(unique_list_of_multipliers)
else:
multiplier = unique_list_of_multipliers[0]
if len(unique_list_of_ccy)>1:
print("%s/%s more than one currency %s" %
(symbol, exchange, str(unique_list_of_multipliers)))
currency = str(unique_list_of_ccy)
else:
currency = unique_list_of_ccy[0]
return [symbol, exchange, currency, multiplier]
import numpy as np
def get_stats_for_data_row(ib, data_row, all_fx_rates):
list_of_contracts = get_list_of_contracts(ib,
data_row.Symbol,
data_row.Exchange,
data_row.Currency,
data_row.Multiplier)
if len(list_of_contracts)==0:
print("Couldn't get list of contracts for %s" % str(data_row))
return missing_data
list_of_market_data = \
get_list_of_market_data_for_list_of_contract_details(ib,
list_of_contract_details=list_of_contracts)
highest_volume_index, contracts_per_day = contract_with_highest_volumes(list_of_market_data)
true_multiplier = adjust_multiplier(data_row.Multiplier,
list_of_contract_details = list_of_contracts,
highest_volume_index = highest_volume_index)
print("Adjusted multiplier from %f to %s" % (float(data_row.Multiplier),
true_multiplier))
dollar_volume = \
normalised_volume_for_list_of_market_data(list_of_market_data=list_of_market_data,
highest_volume_index=highest_volume_index,
multiplier=true_multiplier,
all_fx_rates=all_fx_rates,
currency=data_row.Currency)
risk_per_contract = get_risk_per_contract_in_dollars(list_of_market_data=list_of_market_data,
highest_volume_index=highest_volume_index,
all_fx_rates=all_fx_rates,
multiplier=true_multiplier,
currency=data_row.Currency)
risk_adjusted_cost, slippage = risk_adjusted_cost_and_slippage_per_trade(ib,
list_of_market_data=list_of_market_data,
list_of_contract_details=list_of_contracts,
highest_volume_index=highest_volume_index,
commission=data_row.Commission,
multiplier =true_multiplier
)
pattern = expiry_pattern_string(list_of_market_data=list_of_market_data,
list_of_contract_details=list_of_contracts)
return [data_row.Symbol,
data_row.Exchange,
data_row.Multiplier,
data_row.Currency,
dollar_volume,
risk_adjusted_cost,
risk_per_contract,
contracts_per_day,
true_multiplier,
data_row.Commission,
slippage,
pattern]
def get_list_of_contracts(ib, symbol, exchange, currency, multiplier):
multiplier = float(multiplier)
if multiplier<1.0:
use_multiplier = str(multiplier)
else:
use_multiplier = str(int(multiplier))
future = Future(symbol=symbol,
exchange=exchange,
currency=currency,
multiplier=use_multiplier)
list_of_contract_details = ib.reqContractDetails(future)
list_of_contract_details_valid = remove_expired_contracts(list_of_contract_details)
return list_of_contract_details_valid
def remove_expired_contracts(list_of_contract_details):
list_of_contract_details_valid = [contract_detail
for contract_detail in list_of_contract_details
if contract_still_valid(contract_detail)]
return list_of_contract_details_valid
def contract_still_valid(contract_details):
expiry_as_str = contract_details.contract.lastTradeDateOrContractMonth
expiry_as_date = datetime.datetime.strptime(expiry_as_str, '%Y%m%d')
if expiry_as_date>datetime.datetime.now():
return True
else:
return False
def get_list_of_market_data_for_list_of_contract_details(ib, list_of_contract_details):
list_of_market_data = [ib.reqHistoricalData(contract_details.contract,
durationStr='20 D',
barSizeSetting='1 day',
endDateTime='',
whatToShow='TRADES',
useRTH=True)
for contract_details in list_of_contract_details]
return list_of_market_data
def adjust_multiplier(raw_multiplier,
list_of_contract_details,
highest_volume_index):
raw_multiplier_as_float = float(raw_multiplier)
priceMagnifier = get_price_magnifier(list_of_contract_details,
highest_volume_index=highest_volume_index)
adjusted_multiplier = raw_multiplier_as_float / priceMagnifier
return adjusted_multiplier
def get_price_magnifier(list_of_contract_details,
highest_volume_index):
relevant_contract_details = list_of_contract_details[highest_volume_index]
priceMagnifier = relevant_contract_details.priceMagnifier
return priceMagnifier
def normalised_volume_for_list_of_market_data(list_of_market_data,
highest_volume_index,
multiplier,
currency,
all_fx_rates):
adv_highest_volume = average_volumes_by_contract(list_of_market_data)[highest_volume_index]
daily_stdev = price_standard_deviation_for_contract_with_highest_volume(highest_volume_index,
list_of_market_data)
risk_per_contract_usd = get_risk_per_contract_in_dollars(list_of_market_data=list_of_market_data,
highest_volume_index=highest_volume_index,
multiplier=multiplier,
currency=currency,
all_fx_rates=all_fx_rates)
volume_in_dollar_risk = adv_highest_volume * risk_per_contract_usd
print("Highest volume %f Risk per contract %f Volume in dollar risk %f" %
(adv_highest_volume, daily_stdev, volume_in_dollar_risk))
return volume_in_dollar_risk
def get_risk_per_contract_in_dollars(list_of_market_data,
highest_volume_index,
currency,
all_fx_rates,
multiplier
):
risk_per_contract = get_risk_per_contract(list_of_market_data=list_of_market_data,
highest_volume_index=highest_volume_index,
multiplier=multiplier)
fx_rate = all_fx_rates[currency]
return fx_rate*risk_per_contract
def get_risk_per_contract(list_of_market_data,
highest_volume_index,
multiplier):
daily_stdev = price_standard_deviation_for_contract_with_highest_volume(highest_volume_index=highest_volume_index,
list_of_market_data=list_of_market_data)
risk_per_contract = 16* daily_stdev * multiplier
return risk_per_contract
def risk_adjusted_cost_and_slippage_per_trade(ib,
list_of_contract_details,
list_of_market_data,
highest_volume_index,
multiplier,
commission):
risk_per_contract = get_risk_per_contract(list_of_market_data=list_of_market_data,
multiplier=multiplier,
highest_volume_index=highest_volume_index)
spread = spread_for_contract_with_highest_volume(ib,
highest_volume_index,
list_of_contract_details=list_of_contract_details)
slippage = spread/2
cost_per_trade = commission + multiplier*slippage
cost_per_trade_SR = cost_per_trade / risk_per_contract
print("Risk per contract %f Spread %f Commission %f Cost per trade SR %f"
% (risk_per_contract, spread, commission, cost_per_trade_SR))
return cost_per_trade_SR, slippage
def expiry_pattern_string(list_of_market_data,
list_of_contract_details):
list_of_average_volumes = average_volumes_by_contract(list_of_market_data)
highest_volume = np.nanmax(list_of_average_volumes)
volumes_as_proportion = [volume / highest_volume for volume in list_of_average_volumes]
list_pattern = ["%s:%.2f" % (contract_details.contract.lastTradeDateOrContractMonth,
volume_proportion)
for contract_details, volume_proportion in
zip(list_of_contract_details,
volumes_as_proportion)]
return ",".join(list_pattern)
def contract_with_highest_volumes(list_of_market_data):
list_of_average_volumes = average_volumes_by_contract(list_of_market_data)
highest_volume = max(list_of_average_volumes)
highest_volume_index = list_of_average_volumes.index(highest_volume)
return highest_volume_index, highest_volume
def average_volumes_by_contract(list_of_market_data):
list_of_average_volumes = [volume_for_a_contract(market_data_for_contract)
for market_data_for_contract in list_of_market_data]
return list_of_average_volumes
def price_standard_deviation_for_contract_with_highest_volume(highest_volume_index,
list_of_market_data):
list_of_closing_prices = [daily_data.close for daily_data in list_of_market_data[highest_volume_index]]
std_dev = np.nanstd(list_of_closing_prices)
return std_dev
import datetime
def spread_for_contract_with_highest_volume(ib,
highest_volume_index,
list_of_contract_details):
contract = list_of_contract_details[highest_volume_index].contract
start = ''
end = datetime.datetime.now()
ticks = ib.reqHistoricalTicks(contract, start, end, 1000, 'BID_ASK', useRth=False)
list_of_spreads = [spread_on_tick(tick) for tick in ticks]
return np.nanmedian(list_of_spreads)
def spread_on_tick(tick):
if np.isnan(tick.priceAsk) or np.isnan(tick.priceBid):
return np.nan
if tick.priceAsk<=0:
return np.nan
if tick.priceBid<=0:
return np.nan
if tick.priceBid>tick.priceAsk:
return np.nan
return tick.priceAsk - tick.priceBid
def volume_for_a_contract(market_data_for_contract):
list_of_volumes = [data_for_day.volume for data_for_day in market_data_for_contract]
return np.nanmean(list_of_volumes)
def get_all_fx_rates(ib, new_market_data_pd):
list_of_fx_codes = list(set(list(new_market_data_pd.Currency.values)))
dict_of_codes_and_prices = dict(
[
(code, get_fx_for_code(ib, code))
for code in list_of_fx_codes
]
)
return dict_of_codes_and_prices
def get_fx_for_code(ib, fx_code, invert=False):
if fx_code=="USD":
return 1.0
if invert:
forex = Forex('USD%s' % fx_code)
else:
forex = Forex('%sUSD' % fx_code)
data = ib.reqHistoricalData(forex, durationStr='5 D', barSizeSetting='1 day', endDateTime='', whatToShow='MIDPOINT', useRTH=True)
if len(data)==0:
if invert:
raise Exception("Can't get price for %s" % fx_code)
return get_fx_for_code(ib, fx_code, invert=True)
data_prices = [data_object.close for data_object in data]
avg_price = np.nanmean(data_prices)
if avg_price==0.0:
raise Exception("Couldn't get price for %s" % fx_code)
if invert:
return 1.0/avg_price
return avg_price
ib=IB()
ib.connect('127.0.0.1', 4001, clientId=13)
new_market_data_pd = pd.read_csv('/home/rob/private/projects/new_markets/with_unique_identifers.csv')
all_fx_rates = get_all_fx_rates(ib, new_market_data_pd)
new_market_data_with_stats = []
for row_data in new_market_data_pd.iterrows():
print("Getting data for %s" % str(row_data[1]))
new_data = get_stats_for_data_row(ib, row_data[1], all_fx_rates=all_fx_rates)
if new_data is missing_data:
continue
new_market_data_with_stats.append(new_data)
new_market_data_with_stats_pd = pd.DataFrame(new_market_data_with_stats)
new_market_data_with_stats_pd.columns = ['Symbol', 'Exchange', 'Currency', 'Multiplier',
'Risk Volume $', 'Cost SR', '$ Risk per contract',
'Contracts per day', 'My multiplier',
'Commission', 'Slippage price units',
'Pattern']
new_market_data_with_stats_pd.to_csv('/home/rob/with_stats.csv')
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment