Explore
Groups
No topics yet. Start the conversation.
Description
The below description is supplied in free-text by the user
# %%
from datetime import date, timedelta
import pandas as pd
import sys, inspect
import time
import requests
import io
from novem import Plot
from novem.colors import StaticColor as SC, DynamicColor as DC
from novem.table import Selector as S
# (1/3) DATA COLLECTION
NB_URL = 'https://data.norges-bank.no/api/data/'
NB_URL_OPTIONS = '?format=csv&locale=no'
CURRENCIES = {
'USD': 'US Dollar',
'EUR': 'Euro',
'JPY': 'Japanese Yen',
'GBP': 'British Pound',
'CHF': 'Swiss Franc',
'NOK': 'Norwegian Krone',
'DKK': 'Danish Krone',
'SEK': 'Swedish Krona',
'PLN': 'Polish Zloty',
'CZK': 'Czech Koruna',
'HUF': 'Hungarian Forint',
'ZAR': 'South African Rand',
'TRY': 'Turkish Lira',
'CAD': 'Canadian Dollar',
'MXN': 'Mexican Peso',
'BRL': 'Brazilian Real',
'AUD': 'Australian Dollar',
'NZD': 'New Zealand Dollar',
'SGD': 'Singapore Dollar',
'KRW': 'South Korean Won',
'CNY': 'Chinese Yuan',
'INR': 'Indian Rupee',
'THB': 'Thai Baht'
}
BASE_CCY = ['USD','EUR', 'GBP', 'NOK']
HORIZONS = {'1d': 1, '5d': 5, '1m': 22, '1y': 252}
END_DATE = date.today().isoformat() # Today
START_DATE = (date.today() - timedelta(days=2*365)).isoformat() # Two years ago
WAIT_S = 0.5 # one kind of wait: used both as throttle + retry wait
def fetch_ccy(ccy):
print(f'Fetching {ccy}...')
if ccy == 'NOK':
return pd.DataFrame({'value_date': pd.date_range(start=START_DATE, end=END_DATE),
'ccy': 'NOK', 'NOK': 100, 'UNIT_MULT': 2, 'scale_factor': 100})
url = f'{NB_URL}EXR/B.{ccy}.NOK.SP{NB_URL_OPTIONS}&startPeriod={START_DATE}&endPeriod={END_DATE}'
print(f'URL: {url}')
time.sleep(WAIT_S) # throttle
for _ in range(3): # small retry for transient blocks
r = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=30)
if r.status_code == 200:
df = pd.read_csv(io.StringIO(r.text), sep=';', decimal=',', parse_dates=['TIME_PERIOD'])
df['scale_factor'] = 10**df['UNIT_MULT']
return df.rename(columns={'BASE_CUR': 'ccy', 'OBS_VALUE': 'NOK', 'TIME_PERIOD': 'value_date'})
print(f'HTTP {r.status_code}: {r.text[:120].replace(chr(10)," ")}')
time.sleep(WAIT_S) # same wait, retry
r.raise_for_status()
def switch_base(fx, new_base):
print(f'Switching base to {new_base}...')
if new_base == 'NOK':
# Norwegian's see the world differently. 100 SEK = X NOK
fx = fx[fx['ccy']!=new_base]
fx['ccy_print'] = fx['scale_factor'].astype(str) + ' ' + fx['ccy']
return fx.rename(columns={'NOK': 'value'})
# In the UK and US they see the world differently. 1 GBP = X SEK
fx['ccy_print'] = fx['ccy']
fx_base = fx.loc[fx['ccy'] == new_base, ['value_date', 'NOK']].rename(columns={'NOK': 'NOK_to_base'})
fx = fx.merge(fx_base, on='value_date', how='left')
fx['value'] = fx['NOK_to_base'] * fx['scale_factor'] / fx['NOK']
fx = fx.dropna(subset=['value'])
fx = fx[fx['ccy']!=new_base]
return fx
# (2/3) DATA ENRICHMENT
def enrich_and_take_latest(g, base):
g = g.sort_values('value_date')
latest = g.iloc[-1:].copy() # Latest value per ccy
stdev = g['value'].pct_change().std() # Stdev per ccy
changes = {k: g['value'].pct_change(v).iloc[-1] for k, v in HORIZONS.items()}
zscores = {f'z_{k}': changes[k] / (stdev * (v**0.5)) for k, v in HORIZONS.items()}
if base == 'NOK':
# Do nothing
pass
else:
zscores = {key: -value for key, value in zscores.items()} # Invert z-scores for USD, EUR, GBP etc. (because we want to show strengthening in green, weakening in red)
return latest.assign(**changes, **zscores)
def enhance_fx(fx,base):
fx_latest = (
fx.groupby('ccy', group_keys=False)
.apply(enrich_and_take_latest, base)
.reset_index(drop=True)
)
fx_latest['Description'] = fx_latest['ccy'].map(CURRENCIES)
fx_latest['ccy'] = pd.Categorical(fx_latest['ccy'], categories=list(CURRENCIES.keys()), ordered=True)
fx_latest = fx_latest.sort_values(by='ccy')
fx_latest['ccy'] = fx_latest['ccy'].astype(str)
# 100 and 'SEK' -> 100 SEK
fx_latest = fx_latest.set_index('Description')
def format_number(x: float) -> str:
return f'{x:,.3f}' if x < 10 else f'{x:,.2f}' if x < 100 else f'{x:,.1f}' if x < 1000 else f'{x:,.0f}'
fx_latest['value'] = fx_latest['value'].map(format_number)
return fx_latest
def novem_plot_fx(fx, fx_latest, base):
# (3/3) NOVEM PLOT
visible_cols = ['ccy', 'value'] + list(HORIZONS.keys())
data = fx_latest[visible_cols]
plot_id = 'ccy_moves_vs_' + base.lower()
plt = Plot(plot_id) # Initiate the plot
plt.type = 'table' # Set type to table (waiting to set to mtable until later for faster rendering)
data.rename(columns={'value':base}, inplace=True)
plt.data = data # Pipe the data to novem
print(plt.url) # Show the url of the plot
plt.cell.format = {} # Reset all cell formats
plt.cell.align = '''
: 1: >
: 0 <
0 2: -
'''
border = [0, 4, 7, 12, 15, -1] # Columns to have borders
plt.cell.border = ' '
for b in border:
plt.cell.border += f'{b} : b 1 gray-500' # Add a left border (col = b, row = all, border location = left , width = 1, color = gray-300)
plt.cell.padding= '''
: 1: l 4
: :-2 r 4
0 : t 2
'''
plt.cell.format += S(data.loc[:, '1d':], ',.1%', data)
plt.cell.format += S(data.loc[:, '1m':], ',.0%', data)
# Color each value based on whether it is 'special' i.e. z-score > 1
plt.colors.type = 'ix'
plt.colors = '' # Reset
for horizon in HORIZONS:
z_col = f'z_{horizon}'
col_index = data.columns.get_loc(horizon) + 1
plt.colors += S(fx_latest.loc[fx_latest[z_col] > 1, [horizon]], SC('bg', 'green-100'), data, col_index)
plt.colors += S(fx_latest.loc[fx_latest[z_col] < -1, [horizon]], SC('bg', 'red-100'), data, col_index)
plt.colors += S(fx_latest.loc[fx_latest[z_col] > 1, [horizon]], SC('fg', 'green-600'), data, col_index)
plt.colors += S(fx_latest.loc[fx_latest[z_col] < -1, [horizon]], SC('fg', 'red-600'), data, col_index)
t_latest = max(fx['value_date']).strftime('%d %b %Y')
plt.name = f'Currency moves vs {base} ' + t_latest
plt.title = ''
plt.caption = f'''Global currencies vs {base}. Red (currency weakening) or green (currency strengthening) where move is
greater than standard deviation last two years. Showing
European reference rates set ca. 14:15 CET on {t_latest}. Source: Norges Bank, calculations by novem. '''
plt.type = 'mtable' # Set plot type to mtable at the end (ensures rendering is only done once, faster)
plt.shared += 'public' # Make the graph public
# Include this code in Description
code_block = f'''```python
{inspect.getsource(sys.modules[__name__])}
```'''
plt.description = code_block
# %%
fx_nok = pd.concat([fetch_ccy(ccy) for ccy in CURRENCIES])
for base in BASE_CCY:
fx = switch_base(fx_nok, base)
fx_latest = enhance_fx(fx, base)
novem_plot_fx(fx, fx_latest, base)
```