# Libraries
# ============================================
from sqlalchemy import create_engine, text
import pandas as pd
import plotly.express as px
import requests
import os
# Data Extraction
# ============================================
server = "DESKTOP-FUOV4IE \\ MSSQLSERVER_TAB2"
database = "master"
connection_string = (
f"mssql+pyodbc://@ { server} / { database} "
"?driver=ODBC+Driver+17+for+SQL+Server"
"&trusted_connection=yes"
)
engine = create_engine(connection_string)
with engine.connect () as conn:
# Query h_currencies
result = conn.execute(text("SELECT right(symbol, 3) as symbol, date, rate FROM h_currencies_exh WHERE date > '2001-01-01'" ))
df = pd.DataFrame(result.fetchall(), columns= result.keys())
# Query d_countries
url = "https://raw.githubusercontent.com/guillemmaya92/Analytics/refs/heads/master/Data/DIM_Country.json"
data = pd.read_json(url)
df_countries = pd.DataFrame(data).T.reset_index().rename(columns= {"index" : "cod_iso3" , "ISO2" : "cod_iso2" , "Cod_Currency" : "cod_currency" , "Country_Abr" : "country" })
df_countries['currency' ] = df_countries['Currency' ] + " (" + df_countries['Symbol' ] + ")"
df_countries = df_countries[['cod_currency' , 'currency' , 'cod_iso3' , 'country' ]]
df_countries = df_countries[df_countries['cod_iso3' ] != 'ATA' ]
# Adjust SQS (Somaliland)
sos_lines = [
{"cod_currency" : "SQS" , "currency" : "Somaliland Shilling (S)" , "cod_iso3" : 'SOL' , "country" : "Somaliland" },
]
df_countries = pd.concat([df_countries, pd.DataFrame(sos_lines)], ignore_index= True )
# Data Manipulation
# ============================================
# Format date
df['date' ] = pd.to_datetime(df['date' ])
# Adjust VES (Venezuela)
df['rate' ] = df.apply (lambda row: row['rate' ] / 100000 if row['symbol' ] == 'VEF' else row['rate' ], axis= 1 )
df['symbol' ] = df['symbol' ].apply (lambda x: 'VES' if x == 'VEF' else x)
# Adjust ZMW (Zambia)
df['rate' ] = df.apply (lambda row: row['rate' ] / 1000 if row['symbol' ] == 'ZMK' else row['rate' ], axis= 1 )
df['symbol' ] = df['symbol' ].apply (lambda x: 'ZMW' if x == 'ZMK' else x)
# Adjust USD (United States and others)
usa_lines = [
{"symbol" : "USD" , "date" : "2001-01-01" , "rate" : 1 },
{"symbol" : "USD" , "date" : "2001-05-28" , "rate" : 1 },
]
df = pd.concat([df, pd.DataFrame(usa_lines)], ignore_index= True )
df['date' ] = pd.to_datetime(df['date' ])
# Adjust SSP (South Sudan)
ssp_lines = [
{"symbol" : "SSP" , "date" : "2008-01-01" , "rate" : 3.21 },
{"symbol" : "SSP" , "date" : "2025-05-28" , "rate" : 130 },
]
df = pd.concat([df, pd.DataFrame(ssp_lines)], ignore_index= True )
df['date' ] = pd.to_datetime(df['date' ])
# Adjust SQS (Somaliland)
ssp_lines = [
{"symbol" : "SQS" , "date" : "2001-01-01" , "rate" : 2967 },
{"symbol" : "SQS" , "date" : "2024-01-01" , "rate" : 9703 },
]
df = pd.concat([df, pd.DataFrame(ssp_lines)], ignore_index= True )
df['date' ] = pd.to_datetime(df['date' ])
# Adjust ZWL (Zimbawe)
zwl_lines = [
{"symbol" : "ZWL" , "date" : "2001-01-01" , "rate" : 1 / (10 ** 25 )}
]
df = pd.concat([df, pd.DataFrame(zwl_lines)], ignore_index= True )
df['date' ] = pd.to_datetime(df['date' ])
# Adjust CUP (Cuba)
zwl_lines = [
{"symbol" : "CUP" , "date" : "2001-01-01" , "rate" : 1 }
]
df = pd.concat([df, pd.DataFrame(zwl_lines)], ignore_index= True )
df['date' ] = pd.to_datetime(df['date' ])
# Adjust MRU (Mauritania)
mru_lines = [
{"symbol" : "MRU" , "date" : "2001-01-01" , "rate" : 23.8 }
]
df = pd.concat([df, pd.DataFrame(mru_lines)], ignore_index= True )
df['date' ] = pd.to_datetime(df['date' ])
# Adjust AOA (Angola)
mru_lines = [
{"symbol" : "AOA" , "date" : "2001-01-01" , "rate" : 18.7 }
]
df = pd.concat([df, pd.DataFrame(mru_lines)], ignore_index= True )
df['date' ] = pd.to_datetime(df['date' ])
# Adjust MGA (Madagascar)
mru_lines = [
{"symbol" : "MGA" , "date" : "2001-01-01" , "rate" : 1.27 }
]
df = pd.concat([df, pd.DataFrame(mru_lines)], ignore_index= True )
df['date' ] = pd.to_datetime(df['date' ])
# Get index of min and max date for each symbol
min_idx = df.groupby('symbol' )['date' ].idxmin()
max_idx = df.groupby('symbol' )['date' ].idxmax()
# Select rows and rename columns
df_min = df.loc[min_idx].rename(columns= {'date' : 'min_date' , 'rate' : 'min_rate' })
df_max = df.loc[max_idx].rename(columns= {'date' : 'max_date' , 'rate' : 'max_rate' })
# Merge dataframes minmax
df = pd.merge(df_min[['symbol' , 'min_date' , 'min_rate' ]],
df_max[['symbol' , 'max_date' , 'max_rate' ]],
on= 'symbol' )
# Merge dataframes currencies
df = pd.merge(
df,
df_countries,
how= 'left' ,
left_on= 'symbol' ,
right_on= 'cod_currency'
)
# Filter null currencies
df = df[df['cod_iso3' ].notna()]
# Calculate variation
df['variation' ] = ( ( df['max_rate' ] / df['min_rate' ] ) - 1 )
# Clasification groups
df['class' ] = pd.cut(df['variation' ],
bins= [- float ('inf' ), - 0.5 , - 0.3 , - 0.01 , 0.01 , 0.3 , 0.5 , 5 , 12 , 150 , float ('inf' )],
labels= ['G1' , 'G2' , 'G3' , 'G4' , 'G5' , 'G6' , 'G7' , 'G8' , 'G9' , 'G10' ])
# Classification groups names
class_name_map = {
'G1' : 'Appreciation > 50%' ,
'G2' : 'Appreciation [30-50%]' ,
'G3' : 'Appreciation [1-30%]' ,
'G4' : 'Equal' ,
'G5' : 'Depreciation [1-30%]' ,
'G6' : 'Depreciation [30-50%]' ,
'G7' : 'Depreciation [50-500%]' ,
'G8' : 'Depreciation [500-1000%]' ,
'G9' : 'Depreciation [1000-15000%]' ,
'G10' : 'Depreciation [>15000%]'
}
df['class_name' ] = df['class' ].map (class_name_map)
# Sort by class
df = df.sort_values(by= 'class' )
# Rate and date column
df['before' ] = df['min_rate' ].map (' {:.4f} ' .format ) + " (" + df['min_date' ].dt.strftime('%Y-%m- %d ' ) + ")"
df['after' ] = df['max_rate' ].map (' {:.4f} ' .format ) + " (" + df['max_date' ].dt.strftime('%Y-%m- %d ' ) + ")"
# Download to Excel
download_folder = os.path.join(os.path.expanduser("~" ), "Downloads" )
filename = os.path.join(download_folder, "FIG_EXAPI_Currency_Map.xlsx" )
df.to_excel(filename, index= False )
print (df)
# Data Manipulation
# ============================================
# Map
url = 'https://raw.githubusercontent.com/guillemmaya92/Analytics/refs/heads/master/Maps/world-countries-plotly.geojson'
geojson = requests.get(url).json()
# Custom Map Color
color_map_name = {
'Appreciation > 50%' : '#9CF6D2' ,
'Appreciation [30-50%]' : '#CDFFE8' ,
'Appreciation [1-30%]' : '#EFFFF8' ,
'Equal' : '#FFFFFF' ,
'Depreciation [1-30%]' : '#F2DCDB' ,
'Depreciation [30-50%]' : '#E6B8B7' ,
'Depreciation [50-500%]' : '#DA9694' ,
'Depreciation [500-1000%]' : '#963634' ,
'Depreciation [1000-15000%]' : '#632523' ,
'Depreciation [>15000%]' : "#291312"
}
# Figure map choropleth
fig = px.choropleth(
df,
geojson= geojson,
locations= 'cod_iso3' ,
featureidkey= "properties.iso_a3" ,
color= 'class_name' ,
hover_name= 'country' ,
hover_data= {
'cod_currency' : True ,
'currency' : True ,
'before' : True ,
'after' : True ,
'variation' : ':,.2%' ,
'cod_iso3' : False ,
'class_name' : False
},
color_discrete_map= color_map_name,
category_orders= {'class' : ['G1' , 'G2' , 'G3' , 'G4' , 'G5' , 'G6' , 'G7' , 'G8' , 'G9' , 'G10' ]},
labels= {'class_name' : 'Classification' }
)
# Adjust plot
fig.update_geos(fitbounds= "locations" , visible= False )
fig.update_traces(marker_line_width= 0.5 , marker_line_color= "#808080" )
fig.update_layout(
annotations= [
dict (
text= "<b>The power of a floating exchange rate market</b>" ,
xref= "paper" ,
yref= "paper" ,
x= 0 ,
y= 1.05 ,
showarrow= False ,
font= dict (size= 20 )
),
dict (
text= "While USD is getting stronger, the world are feeling pain" ,
xref= "paper" ,
yref= "paper" ,
x= 0 ,
y= 1.015 ,
showarrow= False ,
font= dict (size= 16 )
),
dict (
text= "(Variation in exchange rates during the period 2001–2025)" ,
xref= "paper" ,
yref= "paper" ,
x= 0 ,
y= 0.985 ,
showarrow= False ,
font= dict (size= 13 )
),
dict (
text= "<b>Data Source:</b> <a href='https://exchangeratesapi.io' target='_blank' style='color:#3c586b;'>exchangerateapi.io</a>" ,
xref= "paper" ,
yref= "paper" ,
x= 0 ,
y= 0 ,
showarrow= False ,
font= dict (size= 12 ),
align= "left"
)
],
legend= dict (
title= dict (text= '<b>Categories</b>' ),
font= dict (size= 11 ),
x= 1.02 ,
y= 0.5 ,
yanchor= 'middle' ,
xanchor= 'left' ,
bgcolor= 'rgba(255,255,255,0.8)' ,
bordercolor= "#808080" ,
borderwidth= 1
)
)
# Save it...
download_folder = os.path.join(os.path.expanduser("~" ), "Downloads" )
filename = os.path.join(download_folder, f"FIG_EXAPI_Currency_Map" )
fig.write_html(filename + ".html" )
# Show the plot!
fig.show()