tschm

import logging
import pandas as pd


class Engine(object):
    def __init__(self, engine):
        self.__engine = engine

    def read(self, table, index_col):
        logging.info('SELECT * FROM {0}'.format(table))

tschm

import numpy as np
import pandas as pd

def drawdown(price):
    high_water_mark = np.empty(len(price.index))
    moving_max_value = 0
    for i, value in enumerate(price.values):
        moving_max_value = max(moving_max_value, value)
        high_water_mark[i] = moving_max_value

tschm

import requests
import logging

class MailHandler(logging.Handler):
    def emit(self, record):
        send_message(subject=self.__subject, text=record, toAdr=self.__toAdr, fromAdr=self.__fromAdr,
                     mailgun=self.__mailgun, mailgunkey=self.__mailgunkey)

    def __init__(self, level, format, subject, toAdr, fromAdr, mailgun, mailgunkey):
        super().__init__()

tschm

def monthlytable(nav, year=1900):
    r = nav.pct_change().dropna()
    # Works better in the first month
    # Compute all the intramonth-returns, instead of reapplying some monthly resampling of the NAV
    return_monthly = r.groupby([lambda x: x.year, lambda x: x.month]).apply(lambda x: (1 + x).prod() - 1.0)
    frame = return_monthly.unstack(level=1).rename(columns=lambda x: calendar.month_abbr[x])
    a = (frame + 1.0).prod(axis=1) - 1.0
    frame["STDev"] = np.sqrt(12) * frame.std(axis=1)
    # make sure that you don't include the column for the STDev in your computation
    frame["YTD"] = a

tschm

rm -rf pybank

# clone the entire github repository. This is kind of inefficient as we delete the .git folder in a moment
git clone [email protected]:lobnek/pybank.git

cd pybank

VERSION=$1
if [ "$VERSION" == "" ] 
then

tschm

def formatter(str="{:0.2f}"):
    return lambda x: str.format(x)


def to_latex(frame, filename, na_rep="", float_format=lambda number: "{:0.2f}".format(number)):
    with open(filename, "w") as file:
        frame.to_latex(buf=file, na_rep=na_rep, float_format=float_format, escape=False)
        

tschm

# Makefile
#

ROOT_DIR      = $(shell dirname $(realpath $(lastword $(MAKEFILE_LIST))))
SPHINXBUILD   = sphinx-build

BUILDDIR      = ${ROOT_DIR}/build
SOURCEDIR     = ${ROOT_DIR}/source

PROJECT       = pylobnek

tschm

# construct a random SO(3) matrix

import numpy as np

# We construct a random element in SO(3)
def rand_so3():
    A = np.random.randn(3,3)
    # normalize the first column
    A[:,0]=A[:,0]/np.linalg.norm(A[:,0], 2)
    # make the 2nd column orthogonal to first column

tschm

import os

import pandas as pd
from influxdb import DataFrameClient


class Client(DataFrameClient):
    def __init__(self, host=None, port=8086, database=None):
        host = host or os.environ["INFLUXDB_HOST"]
        super().__init__(host, port)

tschm

# We create a standard logger with two handlers
# One of them is using io.StringIO for streaming
# Subsequently we add the value() method to the logger, e.g. it will be possible to call
# logger.value() and get the entire history of all messages sent to the logger.
#
# I have never managed to get my head around logging in Python. Any comments are very appreciated.

import logging
import types