BART SIMPSON: Todd, you and Data are Team Strike Force. Nelson, that just leaves you and Martin. MARTIN PRINCE, JR: Team Discovery Channel!
— “Lemon of Troy” (S04E24)
Do I need object orientation for encapsulation in Python? Can useful APIs consist entirely of functions? Should metaclasses ever be used? And, most importantly, when do I actually need to write a class?
Do you…
Then come join us for a session on object orientation!
Object orientation is useful because of its unique features like inheritance and composition right? As it turns out, that’s only part of the story. The usefulness of object orientation in Python does not lie in a mechanical understanding of its features, but in its application to decomposing real world problems.
In this episode, we will cover when and why object orientation should be used in Python code and how it can be used appropriately to simplify your life. Additionally, we will discuss common pitfalls of implementing objects and how you can create concise, usable APIs that shorten your development time and ease end-user scripting.
Keywords
“Why should I even bother?”
print("Let's take a look!")
from sqlite3 import connect
from pathlib import Path
from pandas import read_sql, Categorical, to_datetime, MultiIndex
from numpy import sign
with connect(Path('data') / 'data.db') as con:
trades = (
read_sql('select * from trades', con=con)
.assign(
portfolio=lambda df: df['portfolio'].astype('category'),
ticker=lambda df: df['ticker'].astype('category'),
date=lambda df: to_datetime(df['date']),
)
.set_index(['portfolio', 'date', 'ticker'])
.sort_index()
)
prices = (
read_sql('select * from prices', con=con)
.assign(
ticker=lambda df: Categorical(
df['ticker'],
dtype=trades.index.get_level_values('ticker').dtype,
),
date=lambda df: to_datetime(df['date']),
)
.set_index(['date', 'ticker'])
.sort_index()
)
industries = (
read_sql('select * from industry', con=con)
.assign(
ticker=lambda df: Categorical(
df['ticker'],
dtype=trades.index.get_level_values('ticker').dtype,
),
industry=lambda df: df['industry'].astype('category'),
)
.set_index(['ticker'])
.sort_index()
)
positions = trades.groupby(['portfolio', 'ticker'])['volume'].sum()
direction = sign(positions).map({-1: 'bid', 1: 'ask'}).rename('direction')
market_value = (
prices.loc[
prices.index.get_level_values('date').max()
].stack().rename_axis(['ticker', 'direction']).loc[
MultiIndex.from_arrays([
direction.index.get_level_values('ticker'),
direction.values,
])
].droplevel('direction')
.pipe(lambda s:
positions.to_frame().assign(price=s.values)
)
.product(axis='columns')
)
print(
# trades.head(),
# prices.head(),
# industries.head(),
positions.head(),
direction.head(),
market_value.head(),
sep='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
from sqlite3 import connect
from pathlib import Path
from pandas import read_sql, Categorical, to_datetime, MultiIndex
from numpy import sign
with connect(Path('data') / 'data.db') as con:
trades = (
read_sql('select * from trades', con=con)
.assign(
portfolio=lambda df: df['portfolio'].astype('category'),
ticker=lambda df: df['ticker'].astype('category'),
date=lambda df: to_datetime(df['date']),
)
.set_index(['portfolio', 'date', 'ticker'])
.sort_index()
)
prices = (
read_sql('select * from prices', con=con)
.assign(
ticker=lambda df: Categorical(
df['ticker'],
dtype=trades.index.get_level_values('ticker').dtype,
),
date=lambda df: to_datetime(df['date']),
)
.set_index(['date', 'ticker'])
.sort_index()
)
industries = (
read_sql('select * from industry', con=con)
.assign(
ticker=lambda df: Categorical(
df['ticker'],
dtype=trades.index.get_level_values('ticker').dtype,
),
industry=lambda df: df['industry'].astype('category'),
)
.set_index(['ticker'])
.sort_index()
)
positions = trades.groupby(['portfolio', 'ticker'])['volume'].sum()
direction = sign(positions).map({-1: 'bid', 1: 'ask'}).rename('direction')
market_value = (
prices.loc[
prices.index.get_level_values('date').max()
].stack().rename_axis(['ticker', 'direction']).loc[
MultiIndex.from_arrays([
direction.index.get_level_values('ticker'),
direction.values,
])
].droplevel('direction')
.pipe(lambda s:
positions.to_frame().assign(price=s.values)
)
.product(axis='columns')
)
included = industries.loc[lambda s: s.isin(['tech', 'finance']).values]
excluded = industries.loc[lambda s: ~s.isin(['tech', 'finance']).values]
print(
# industries,
# industries.loc[lambda s: s.isin(['tech', 'finance']).values],
included,
excluded,
sep='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
from sqlite3 import connect
from pathlib import Path
from pandas import read_sql, Categorical, to_datetime, MultiIndex, IndexSlice
from numpy import sign
with connect(Path('data') / 'data.db') as con:
trades = (
read_sql('select * from trades', con=con)
.assign(
portfolio=lambda df: df['portfolio'].astype('category'),
ticker=lambda df: df['ticker'].astype('category'),
date=lambda df: to_datetime(df['date']),
)
.set_index(['portfolio', 'date', 'ticker'])
.sort_index()
)
prices = (
read_sql('select * from prices', con=con)
.assign(
ticker=lambda df: Categorical(
df['ticker'],
dtype=trades.index.get_level_values('ticker').dtype,
),
date=lambda df: to_datetime(df['date']),
)
.set_index(['date', 'ticker'])
.sort_index()
)
industries = (
read_sql('select * from industry', con=con)
.assign(
ticker=lambda df: Categorical(
df['ticker'],
dtype=trades.index.get_level_values('ticker').dtype,
),
industry=lambda df: df['industry'].astype('category'),
)
.set_index(['ticker'])
.sort_index()
)
included = industries.loc[lambda s: s.isin(['tech', 'finance']).values].index
excluded = industries.loc[lambda s: ~s.isin(['tech', 'finance']).values].index
included_positions = trades.groupby(['portfolio', 'ticker'])['volume'].sum().loc[IndexSlice[:, included, :]]
excluded_positions = trades.groupby(['portfolio', 'ticker'])['volume'].sum().loc[IndexSlice[:, excluded, :]]
included_direction = sign(included_positions).map({-1: 'bid', 1: 'ask'}).rename('direction')
excluded_direction = sign(excluded_positions).map({-1: 'bid', 1: 'ask'}).rename('direction')
included_market_value = (
prices.loc[
prices.index.get_level_values('date').max()
].stack().rename_axis(['ticker', 'direction']).loc[
MultiIndex.from_arrays([
included_direction.index.get_level_values('ticker'),
included_direction.values,
])
].droplevel('direction')
.pipe(lambda s:
included_positions.to_frame().assign(price=s.values)
)
.product(axis='columns')
)
excluded_market_value = (
prices.loc[
prices.index.get_level_values('date').max()
].stack().rename_axis(['ticker', 'direction']).loc[
MultiIndex.from_arrays([
excluded_direction.index.get_level_values('ticker'),
excluded_direction.values,
])
].droplevel('direction')
.pipe(lambda s:
excluded_positions.to_frame().assign(price=s.values)
)
.product(axis='columns')
)
print(
# included,
# excluded,
# included_positions,
# excluded_positions,
included_market_value,
excluded_market_value,
sep='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
print("Let's take a look!")
class T:
def __init__(self, value):
self.value = value
obj1 = T(123)
obj2 = T(456)
class T:
def __init__(self, value):
self._value = value
def set_value(self, new_value):
self._value = new_value
def get_value(self):
return self._value
obj1 = T(123)
obj2 = T(456)
print(
f'{obj1.get_value() = }',
sep='\n', end='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
obj1.set_value(777)
print(
f'{obj1.get_value() = }',
f'{obj2.get_value() = }',
sep='\n', end='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
class T:
def __init__(self, value):
self._value = value
@property
def value(self):
return self._value
@value.setter
def value(self, new_value):
self._value = new_value
obj1 = T(123)
obj2 = T(456)
print(
f'{obj1.value = }',
sep='\n', end='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
obj1.value = 777
print(
f'{obj1.value = }',
sep='\n', end='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
class T:
def __init__(self, state):
self.state = state
def inc(self):
self.state += 1
obj = T(123)
print(
f'{obj.state = }',
sep='\n', end='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
obj.inc()
print(
f'{obj.state = }',
sep='\n', end='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
def create_closure(state):
def get():
return state
def inc():
nonlocal state
state += 1
return get, inc
obj = create_closure(123)
print(
f'{obj[0]() = }',
sep='\n', end='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
obj[1]()
print(
f'{obj[0]() = }',
sep='\n', end='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
from collections import namedtuple
def create_closure(state):
def get():
return state
def inc():
nonlocal state
state += 1
return namedtuple('methods', 'state inc')(get, inc)
obj = create_closure(123)
print(
f'{obj.state() = }',
sep='\n', end='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
obj.inc()
print(
f'{obj.state() = }',
sep='\n', end='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
class Api:
def first(self, state):
self.state = state
def second(self, mode):
self.mode = mode
return self.state + 1
def third(self):
return self.state * 2 if self.mode else self.state * 3
a = Api()
print(
f'{a.first(123) = }',
f'{a.second(True) = }',
f'{a.third() = }',
sep='\n',
)
class Api:
def first(self): print('first')
def second(self): print('second')
def third(self): print('third')
a = Api()
a.first()
a.third()
a.second()
def api():
print('first')
...
print('second')
...
print('third')
api()
def api():
print('first')
yield
print('second')
yield ...
print('third')
yield
a = api()
next(a)
...
print(f'{next(a) = }')
...
next(a)
def api():
print('first')
state = yield
print('second')
mode = yield state + 1
print('third')
yield state * 2 if mode else state * 3
a = api()
next(a)
print(f'{a.send(123) = }')
print(f'{a.send(True) = }')
print("Let's take a look!")
from pandas import DataFrame
df = (
DataFrame(...)
.set_index(...)
.sort_index()
.assign(...)
.loc[...]
.groupby(...).mean()
)
from fake_pandas import set_index, sort_index, assign, loc, groupby, mean
df = { ... }
df = set_index(df, ...)
df = sort_index(df)
df = assign(df, ...) # df[...] = ...
groupby(df, ..., mean)
from collections import Counter
c1 = Counter({'a': 3, 'b': 2, 'c': 1})
c2 = Counter({'a': 5, 'b': 3, 'd': 2})
print(
c1,
c2,
c1 + c2,
sep='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
from pandas import Series
s1 = Series({'a': 3, 'b': 2, 'c': 1})
s2 = Series({'a': 5, 'b': 3, 'd': 2})
print(
s1,
s2,
s1 + s2,
sep='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
from collections import Counter
from pandas import Series
x, y = 123, 456
x, y = 1.23, 4.56
x, y = 123+456j, 456+789j
x, y = Counter({'a': 1, 'b': 2, 'c': 3}), Counter({'a': 4, 'b': 5, 'c': 6})
x, y = Series({'a': 1, 'b': 2, 'c': 3}), Series({'a': 4, 'b': 5, 'c': 6})
print(
f'{x + y = }',
sep='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
from functools import reduce
from operator import or_
from numpy.random import default_rng
rng = default_rng(0)
xs = rng.normal(size=100)
masks = [
xs > 0,
xs.astype(int) % 2 == 0,
]
print(
# xs[(xs > 0) | (xs.astype(int) % 2 == 0)],
xs[reduce(or_, masks)]
sep='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
class T:
__add__ = lambda *_: None
__sub__ = lambda *_: None
__mul__ = lambda *_: None
__pow__ = lambda *_: None
__truediv__ = lambda *_: None
__floordiv__ = lambda *_: None
x, y = T(), T()
x + y
x - y
x * y
x / y
x // y
import math
__import__
class T:
pass
__build_class__
xs = ..., ...
for x in xs: pass
xi = iter(xs)
while True:
try:
x = next(xi)
except StopIteration:
break
class T:
def __iter__(self):
return self
def __next__(self):
raise StopIteration()
for _ in T(): pass
d = {(k := ...): (v := ...)}
d[k]
d[k] = v
d.__getitem__(k)
d.__setitem__(k, v)
from collections import defaultdict
from collections import Counter
from collections import ChainMap
d = defaultdict(int, {(k := ...): 0})
d[k]
d[...]
d = Counter({(k := ...): 0})
d[k]
d[...]
d = ChainMap({}, {(k := ...): ...})
d[k]
from contextlib import nullcontext
with nullcontext() as ctx:
pass
ctxmgr = nullcontext()
ctx = ctxmgr.__enter__()
try:
...
except Exception:
ctxmgr.__exit__(..., ..., ...)
else:
ctxmgr.__exit__()
class Context:
def __enter__(self):
print('before')
def __exit__(self, exc_type, exc_value, traceback):
print('after, error' if exc_type is not None else 'after, no error')
with Context():
...
from contextlib import contextmanager
@contextmanager
def context():
print('before')
try:
yield
except Exception:
print('after, error')
else:
print('after, no error')
with context():
...
from collections import namedtuple
T = namedtuple('T', 'a b c')
obj = T(1, 2, 3)
from collections import namedtuple
T = namedtuple('T', 'a b c', verbose=True)
from dataclasses import dataclass
@dataclass
class T:
a : int
b : int
c : int
obj = T(1, 2, 3)
from enum import Enum
T = Enum('T', 'A B C')
T.A, T.B, T.C
x = yx ≠ yx > yx < yx ≤ yx ≥ yfrom functools import total_ordering
@total_ordering
class T:
__eq__ = lambda self, other: None
__lt__ = lambda self, other: None
print("Let's take a look!")
if __name__ == '__main__':
trading_data = ...
other_trading_data = ...
if __name__ == '__main__':
trading_data = {
'base': ...,
'other': ...,
}
subplots_mosaic(
trading_data.keys()
)
for k, td in trading_data.items()
axes[k] = td.plot()
from sqlite3 import connect
from pathlib import Path
from pandas import read_sql
if __name__ == '__main__':
with connect(Path('data') / 'data.db') as con:
trading_data = read_sql(..., con=con)
if __name__ == '__main__':
trading_data = TradingData(...)
class TradingData:
def __init__(self, query):
...
class TradingData:
def __init__(self, raw_data):
...
def load_from_query(query):
return TradingData(...)
class TradingData:
def __init__(self, raw_data):
...
@classmethod
def load_from_query(cls, query):
return cls(...)
from dataclasses import dataclass
from pandas import DataFrame, Series, read_sql, to_datetime, Categorical
from pathlib import Path
from sqlite3 import connect
@dataclass(frozen=True)
class TradingData:
trades : DataFrame
prices : Series
industries : Series
@classmethod
def from_db(cls, *, con):
trades = (
read_sql('select * from trades', con=con)
.assign(
portfolio=lambda df: df['portfolio'].astype('category'),
ticker=lambda df: df['ticker'].astype('category'),
date=lambda df: to_datetime(df['date']),
)
.set_index(['portfolio', 'date', 'ticker'])
.sort_index()
)
prices = (
read_sql('select * from prices', con=con)
.assign(
ticker=lambda df: Categorical(
df['ticker'],
dtype=trades.index.get_level_values('ticker').dtype,
),
date=lambda df: to_datetime(df['date']),
)
.set_index(['date', 'ticker'])
.sort_index()
)
industries = (
read_sql('select * from industry', con=con)
.assign(
ticker=lambda df: Categorical(
df['ticker'],
dtype=trades.index.get_level_values('ticker').dtype,
),
industry=lambda df: df['industry'].astype('category'),
)
.set_index(['ticker'])
.sort_index()
)
return cls(trades=trades, prices=prices, industries=industries)
if __name__ == '__main__':
with connect(Path('data') / 'data.db') as con:
trading_data = TradingData.from_db(con=con)
print(
trading_data.trades.head(3),
trading_data.prices.head(3),
trading_data.trades.groupby(['portfolio', 'ticker']).sum().head(3),
sep='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
from dataclasses import dataclass
from pandas import DataFrame, Series, read_sql, to_datetime, Categorical, MultiIndex
from numpy import sign
from pathlib import Path
from sqlite3 import connect
from functools import cached_property
@dataclass(frozen=True)
class TradingData:
trades : DataFrame
prices : Series
industries : Series
@classmethod
def from_db(cls, *, con):
trades = (
read_sql('select * from trades', con=con)
.assign(
portfolio=lambda df: df['portfolio'].astype('category'),
ticker=lambda df: df['ticker'].astype('category'),
date=lambda df: to_datetime(df['date']),
)
.set_index(['portfolio', 'date', 'ticker'])
.sort_index()
)
prices = (
read_sql('select * from prices', con=con)
.assign(
ticker=lambda df: Categorical(
df['ticker'],
dtype=trades.index.get_level_values('ticker').dtype,
),
date=lambda df: to_datetime(df['date']),
)
.set_index(['date', 'ticker'])
.sort_index()
)
industries = (
read_sql('select * from industry', con=con)
.assign(
ticker=lambda df: Categorical(
df['ticker'],
dtype=trades.index.get_level_values('ticker').dtype,
),
industry=lambda df: df['industry'].astype('category'),
)
.set_index(['ticker'])
.sort_index()
)
return cls(trades=trades, prices=prices, industries=industries)
@cached_property
def positions(self):
positions = self.trades.groupby(['portfolio', 'ticker'])['volume'].sum()
direction = sign(positions).map({-1: 'bid', 1: 'ask'}).rename('direction')
market_value = (
self.prices.loc[
self.prices.index.get_level_values('date').max()
].stack().rename_axis(['ticker', 'direction']).loc[
MultiIndex.from_arrays([
direction.index.get_level_values('ticker'),
direction.values,
])
].droplevel('direction')
.pipe(lambda s:
positions.to_frame().assign(price=s.values)
)
.product(axis='columns')
.rename('market value')
)
return positions.to_frame().join(market_value)
if __name__ == '__main__':
with connect(Path('data') / 'data.db') as con:
trading_data = TradingData.from_db(con=con)
print(
# trading_data.trades.head(3),
# trading_data.prices.head(3),
trading_data.positions.head(3),
sep='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
from dataclasses import dataclass
from pandas import DataFrame, Series, read_sql, to_datetime, Categorical, MultiIndex
from numpy import sign
from pathlib import Path
from sqlite3 import connect
from functools import cached_property
@dataclass(frozen=True)
class TradingData:
trades : DataFrame
prices : Series
industries : Series
@classmethod
def from_db(cls, *, con):
trades = (
read_sql('select * from trades', con=con)
.assign(
portfolio=lambda df: df['portfolio'].astype('category'),
ticker=lambda df: df['ticker'].astype('category'),
date=lambda df: to_datetime(df['date']),
)
.set_index(['portfolio', 'date', 'ticker'])
.sort_index()
)
prices = (
read_sql('select * from prices', con=con)
.assign(
ticker=lambda df: Categorical(
df['ticker'],
dtype=trades.index.get_level_values('ticker').dtype,
),
date=lambda df: to_datetime(df['date']),
)
.set_index(['date', 'ticker'])
.sort_index()
)
industries = (
read_sql('select * from industry', con=con)
.assign(
ticker=lambda df: Categorical(
df['ticker'],
dtype=trades.index.get_level_values('ticker').dtype,
),
industry=lambda df: df['industry'].astype('category'),
)
.set_index(['ticker'])
.sort_index()
)
return cls(trades=trades, prices=prices, industries=industries)
@cached_property
def positions(self):
positions = self.trades.groupby(['portfolio', 'ticker'])['volume'].cumsum()
direction = sign(positions).map({-1: 'bid', 1: 'ask'}).rename('direction')
market_value = (
self.prices.loc[
self.prices.index.get_level_values('date').max()
].stack().rename_axis(['ticker', 'direction']).loc[
MultiIndex.from_arrays([
direction.index.get_level_values('ticker'),
direction.values,
])
].droplevel('direction')
.pipe(lambda s:
positions.to_frame().assign(price=s.values)
)
.product(axis='columns')
.rename('market value')
)
return positions.to_frame().join(market_value)
if __name__ == '__main__':
with connect(Path('data') / 'data.db') as con:
trading_data = TradingData.from_db(con=con)
print(
# trading_data.trades.head(3),
# trading_data.prices.head(3),
trading_data.positions.head(3),
sep='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
from dataclasses import dataclass
from pandas import DataFrame, Series, read_sql, to_datetime, Categorical, MultiIndex
from numpy import sign
from pathlib import Path
from sqlite3 import connect
from functools import cached_property
@dataclass(frozen=True)
class TradingData:
trades : DataFrame
prices : Series
industries : Series
@classmethod
def from_db(cls, *, con):
trades = (
read_sql('select * from trades', con=con)
.assign(
portfolio=lambda df: df['portfolio'].astype('category'),
ticker=lambda df: df['ticker'].astype('category'),
date=lambda df: to_datetime(df['date']),
)
.set_index(['portfolio', 'date', 'ticker'])
.sort_index()
)
prices = (
read_sql('select * from prices', con=con)
.assign(
ticker=lambda df: Categorical(
df['ticker'],
dtype=trades.index.get_level_values('ticker').dtype,
),
date=lambda df: to_datetime(df['date']),
)
.set_index(['date', 'ticker'])
.sort_index()
)
industries = (
read_sql('select * from industry', con=con)
.assign(
ticker=lambda df: Categorical(
df['ticker'],
dtype=trades.index.get_level_values('ticker').dtype,
),
industry=lambda df: df['industry'].astype('category'),
)
.set_index(['ticker'])
.sort_index()
)
return cls(trades=trades, prices=prices, industries=industries)
@cached_property
def positions(self):
return self.trades.groupby(['portfolio', 'ticker'])['volume'].cumsum()
@cached_property
def revals(self):
direction = sign(self.positions).map({-1: 'bid', 1: 'ask'}).rename('direction')
return (
self.prices.stack()
.rename_axis(['date', 'ticker', 'direction']).rename('price')
.loc[
MultiIndex.from_arrays([
direction.index.get_level_values('date').floor('D'),
direction.index.get_level_values('ticker'),
direction.values,
])
]
.droplevel('direction')
.pipe(lambda s:
self.positions.to_frame().assign(price=s.values)
)
)
if __name__ == '__main__':
with connect(Path('data') / 'data.db') as con:
trading_data = TradingData.from_db(con=con)
for sh in shocks:
with trading_data.shock_ir(sh.ir) as td:
td.position
with td.shock_ir(sh.rr) as td1:
td1.position
with td1.shock_r(sh.r) as td2:
td2.position
with td.shock_r(sh.r) as td1:
td1.position
with td.shock_ir(sh.rr) as td1:
td1.position
print(
# trading_data.trades.head(3),
# trading_data.prices.head(3),
trading_data.positions.head(3),
trading_data.revals,
sep='\n{}\n'.format('\N{box drawings light horizontal}' * 40),
)
from pandas import date_range, MultiIndex, Series
from numpy import triu_indices
dates = date_range('2020-01-01', periods=3)
idx = MultiIndex.from_product([
dates,
dates,
])
mask = triu_indices(len(dates))
print(
idx,
mask,
MultiIndex.from_arrays([dates[idx] for idx in mask]),
sep='\n',
)