Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Gist with additional files from For Python Quants Bootcamp, May 2017, New York City
3 + 4
3 * 4
3 / 4
type(3)
type(4)
3 ** 4
sqrt(3)
3 ** 0.5
import math
math.sqrt(3)
print("Python.")
a = 3
b = 0.75
c = 'Python.'
d = "He said:'I am late.'"
d
d = "He said:"I am late.""
d = 'He said:"I am late."'
d
a
print(a)
a
b
a * b
a ** b
d
2 * d
d + d
d + d * 2
d / d
d / 2
d
d[0]
d[1]
len(d)
d[20]
d[19]
d[-1]
d[-2]
d[-20]
d[-21]
d[2]
d[:2]
d[:2] + d[2:]
d[2:]
d[2:7]
d[2:7:2]
d[::2]
d[::-1]
range(10)
type(range(10))
for i in range(10):
print('For Python Quants')
for i in range(10):
print(i)
for i in range(10):
print(i ** 2)
%magic
%lsmagic
%hist
%hist?
%history?
len?
for i in range(10):
print(d[i])
for c in d:
print(c)
for _ in d:
print(_)
c
for _ in d:
print(_, end='')
for _ in d:
print(_, end='|')
for x in range(10):
print(x)
for x in range(10):
print(x ** 2)
l = [x for x in range(10)]
l
l = [x ** 2 for x in range(10)]
l
type(l)
l2 = [x ** 2 for x in range(10) if x > 2]
l2
l2 = [x ** 2 for x in range(10) if (x > 2) and (x < 8)]
l2
l[0]
l[:5]
l[5:]
l[::-1]
l = [x ** 2 for x in range(10)]
10 % 2
11 % 2
l = [x ** 2 for x in range(10) if x % 2 == 0]
l
l = [x for x in range(20) if x % 2 == 0]
l
for x in range(20):
for y in range(10, 50):
if x % 2 == 0:
# then do something
pass
def f(x):
return x ** 2
f
f(10)
f(10.5)
l = [f(x) for x in range(20) if x % 2 == 0]
l
l3 = [5, 'fpq', a, l]
l3
l3.append('this is new')
l3
l3.append(f)
l3
l3[-1](5)
l.append('new')
l
l3
l
l3
def is_prime(I):
for i in range(2, I):
if I % i == 0:
return False
return True
is_prime(8)
is_prime(10)
is_prime(11)
is_prime(13)
l = [is_prime(x) for x in range(2, 101)]
l
l = [is_prime(x) for x in range(2, 20)]
l
class MyClass(object):
pass
class my_class(object):
pass
int(Ture)
int(True)
int(False)
while True:
print('hi')
while 2:
print('hi')
2 == 2
True == 2
True == 1
def is_prime_2(I):
for i in range(2, I ** 0.5):
if I % i == 0:
return False
return True
is_prime_2(10)
def is_prime_2(I):
for i in range(2, int(I ** 0.5)):
if I % i == 0:
return False
return True
int(2.3)
int(2.7)
def is_prime_2(I):
for i in range(2, int(I ** 0.5) + 1):
if I % i == 0:
return False
return True
is_prime_2(10)
is_prime_2(11)
%ed
p1 = int(1e8 + 1)
p2 = int(1e8 + 3)
p1
p2
is_prime(p1)
is_prime(p2)
p2 = 2** 17 − 1
p2 = 2 ** 17 - 1
p2
p2 = 2 ** 31 - 1
p2
%time is_prime(p1)
%time is_prime(p2)
p2 = 2 ** 17 - 1
%time is_prime(p2)
%time is_prime_2(p2)
%time is_prime_2(int(2**31 - 1))
def is_prime_3(I):
if I % 2 == 0:
return False
for i in range(3, int(I ** 0.5) + 1, 2):
if I % i == 0:
return False
return True
%time is_prime_3(int(2**31 - 1))
%ed is_prime_3
%ed -p
from math import sqrt
sqrt(4)
ls
cd ..
ls
cd bc
!mkdir bc
cd bc/
%hist -f bc_day_1_section_02
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
#
# Tick Data Client
# with ZeroMQ
#
import zmq
import datetime
context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect('tcp://127.0.0.1:5555')
socket.setsockopt_string(zmq.SUBSCRIBE, '')
while True:
msg = socket.recv_string()
t = datetime.datetime.now()
print(str(t) + ' | ' + msg)
#
# Tick Data Client
# with ZeroMQ
#
import zmq
import datetime
import plotly.plotly as ply
import plotly.tools as tls
from plotly.graph_objs import *
stream_ids = tls.get_credentials_file()['stream_ids']
# socket
context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect('tcp://127.0.0.1:5555')
socket.setsockopt_string(zmq.SUBSCRIBE, '')
# plotting
s = Stream(maxpoints=100, token=stream_ids[0])
t = Scatter(x=[], y=[], name='tick data', mode='lines+markers', stream=s)
d = Data([t])
l = Layout(title='Bootcamp Tick Data')
f = Figure(data=d, layout=l)
ply.plot(f, filename='fpq_bootcamp', auto_open=True)
st = ply.Stream(stream_ids[0])
st.open()
while True:
msg = socket.recv_string()
t = datetime.datetime.now()
sym, value = msg.split()
print(str(t) + ' | ' + msg)
st.write({'x': t, 'y': float(value)})
#
# Tick Data Server
# with ZeroMQ
#
import zmq
import time
import random
context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind('tcp://127.0.0.1:5555')
AMZN = 100.
while True:
AMZN += random.gauss(0, 1) * 0.5
msg = 'AMZN %s' % AMZN
socket.send_string(msg)
print(msg)
time.sleep(random.random() * 2)
#
# tpqoa is a wrapper class for the
# Oanda v20 API (RESTful & streaming)
# (c) Dr. Yves J. Hilpisch
# The Python Quants GmbH
#
import v20
import pandas as pd
import datetime as dt
import configparser
class tpqoa(object):
''' tpqoa is a Python wrapper class for the Oanda v20 API. '''
def __init__(self, conf_file):
''' Init function expecting a configuration file with
the following content:
[oanda_v20]
account_id = XYZ-ABC-...
access_token = ZYXCAB...
Parameters
==========
conf_file: string
path to and filename of the configuration file, e.g. '/home/me/oanda.cfg'
'''
self.config = configparser.ConfigParser()
self.config.read(conf_file)
self.access_token = self.config['oanda_v20']['access_token']
self.account_id = self.config['oanda_v20']['account_id']
self.ctx = v20.Context(
hostname='api-fxpractice.oanda.com',
port=443,
ssl=True,
application='sample_code',
token=self.access_token,
datetime_format='RFC3339')
self.ctx_stream = v20.Context(
hostname='stream-fxpractice.oanda.com',
port=443,
ssl=True,
application='sample_code',
token=self.access_token,
datetime_format='RFC3339'
)
self.suffix = '.000000000Z'
def get_instruments(self):
''' Retrieves and returns all instruments for the given account. '''
resp = self.ctx.account.instruments(self.account_id)
instruments = resp.get('instruments')
instruments = [ins.dict() for ins in instruments]
instruments = [(ins['displayName'], ins['name'])
for ins in instruments]
return instruments
def transform_datetime(self, dt):
''' Transforms Python datetime object to string. '''
if isinstance(dt, str):
dt = pd.Timestamp(dt).to_pydatetime()
return dt.isoformat('T') + self.suffix
def get_history(self, instrument, start, end,
granularity, price):
''' Retrieves historical data for instrument.
Parameters
==========
instrument: string
valid instrument name
start, end: datetime, str
Python datetime or string objects for start and end
granularity: string
a string like 'S5', 'M1' or 'D'
price: string
one of 'A' (ask) or 'B' (bid)
Returns
=======
data: pd.DataFrame
pandas DataFrame object with data
'''
start = self.transform_datetime(start)
end = self.transform_datetime(end)
raw = self.ctx.instrument.candles(
instrument=instrument,
fromTime=start, toTime=end,
granularity=granularity, price=price)
raw = raw.get('candles')
raw = [cs.dict() for cs in raw]
for cs in raw:
cs.update(cs['ask'])
del cs['ask']
if len(raw) == 0:
return 'No data available.'
data = pd.DataFrame(raw)
data['time'] = pd.to_datetime(data['time'])
data = data.set_index('time')
data.index = pd.DatetimeIndex(data.index)
for col in list('ohlc'):
data[col] = data[col].astype(float)
return data
def create_order(self, instrument, units):
''' Places order with Oanda.
Parameters
==========
instrument: string
valid instrument name
units: int
number of units of instrument to be bought (positive int, eg 'units=50')
or to be sold (negative int, eg 'units=-100')
'''
request = self.ctx.order.market(
self.account_id,
instrument=instrument,
units=units,
)
order = request.get('orderFillTransaction')
print('\n\n', order.dict(), '\n')
def stream_data(self, instrument, stop=None):
''' Starts a real-time data stream.
Parameters
==========
instrument: string
valid instrument name
'''
self.stream_instrument = instrument
self.ticks = 0
response = self.ctx_stream.pricing.stream(
self.account_id, snapshot=True,
instruments=instrument)
for msg_type, msg in response.parts():
# print(msg_type, msg)
if msg_type == 'pricing.Price':
self.ticks +=1
self.on_success(msg.time,
float(msg.bids[0].price),
float(msg.asks[0].price))
if stop is not None:
if self.ticks >= stop:
break
def on_success(self, time, bid, ask):
''' Method called when new data is retrieved. '''
print(time, bid, ask)
def get_account_summary(self, detailed=False):
''' Returns summary data for Oanda account.'''
if detailed is True:
response = self.ctx.account.get(self.account_id)
else:
response = self.ctx.account.summary(self.account_id)
raw = response.get('account')
return raw.dict()
def get_transactions(self, tid=0):
''' Retrieves and returns transactions data. '''
response = self.ctx.transaction.since(self.account_id, id=tid)
transactions = response.get('transactions')
transactions = [t.dict() for t in transactions]
return transactions
def print_transactions(self, tid=0):
''' Prints basic transactions data. '''
transactions = self.get_transactions(tid)
for trans in transactions:
templ = '%5s | %s | %9s | %12s'
print(templ % (trans['id'],
trans['time'],
trans['instrument'],
trans['units']))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.