Developing the Attribution Report
Sector Attribution
In this example we look at how to create the Sector Attribution report.
Do note that we will be using the same Data Block that we created for the Top/Bottom N Holdings example.
Below guideline shows how to create the Data Object and fabricate the Asset Allocation Pie Chart.
Below is the Component Inventory for the Attribution Report. (Layers “B” and “C”).
Layers “A” of the Component Inventory for the Attribution Report.
Attribution data extract block with 2 dependency blocks.
Block Name: DEMO_SnowFlakeAttributionDetailsExtract
Output Type: Data Table
Category: Interface
Block Type: Snowflake Db Call
Definition:
{
"type": "table",
"sql": "SELECT * from DEMO_DB.DBO.ATTRIBUTION WHERE PORTFOLIOCODE = ? AND HISTORYDATE = ? AND CATEGORY=? AND CATEGORYSCHEME = ? AND WITHCASH = ? AND CURRENCYCODE = ? AND PERIOD = ?",
"parameters": ["AccountCode", "AsofDate","Category","CategoryScheme","WithCash","CurrencyCode","Period"]
}
Meta Data:
Request Parameters:
Dependency:
Attribution data transform block:
Block Name: DEMO_AttributionCategoryDetails
Output Type: Table
Category: Transform
Block Type: Python
Definition:
# Parameters for retrieving Attribution data
AccountCode=""
AsofDate=""
CurrencyCode=""
ShowComposite=""
Purpose=""
WithCash=""
CategoryScheme=""
Category=""
Period= ""
ShowCash=""
if(params.__len__()>0):
AccountCode = params["AccountCode"]
AsofDate = params["AsofDate"]
if("CurrencyCode" in params.keys()):
CurrencyCode = params["CurrencyCode"]
if("ShowComposite" in params.keys()):
ShowComposite = params["ShowComposite"]
if("Purpose" in params.keys()):
Purpose = params["Purpose"]
if("CategoryScheme" in params.keys()):
CategoryScheme = params["CategoryScheme"]
if("Category" in params.keys()):
Category = params["Category"]
if("WithCash" in params.keys()):
WithCash = params["WithCash"]
if("ShowCash" in params.keys()):
ShowCash = params["ShowCash"]
if("Period" in params.keys()):
Period = params["Period"]
#Getting accountcode
AccountCode = read("DEMO_AccountbyPurpose",{"AccountCode":AccountCode,"Purpose":Purpose,"ShowComposite":ShowComposite})["values"]
if CurrencyCode.lower() == "default":
setCurrency= read("DEMO_GetDefaultCurruncyCode", {"AccountCode":AccountCode})["values"]
else:
setCurrency=CurrencyCode
# Read data block to retrieve Attribution data
detailsdata_df = pd.DataFrame(read("DEMO_SnowFlakeAttributionDetailsExtract",{"AccountCode":AccountCode,"AsofDate":AsofDate,"CategoryScheme":CategoryScheme,"Category":Category,"WithCash":WithCash,"CurrencyCode":setCurrency,"Period":Period})["data"])
totaldata_df= pd.DataFrame(read("DEMO_SnowFlakeAttributionDetailsExtract",{"AccountCode":AccountCode,"AsofDate":AsofDate,"CategoryScheme":CategoryScheme,"Category":"Total","WithCash":WithCash,"CurrencyCode":setCurrency,"Period":Period})["data"])
data_df = pd.concat([detailsdata_df, totaldata_df], axis=0, ignore_index=True)
CategoryLevelData_df = data_df.loc[data_df['SECONDARYCATEGORY'] .isnull()].reset_index(drop=True)
final_result_list = []
transform_column_map = {
"PORTFOLIOCODE":"PortfolioCode",
"CATEGORYSCHEME":"CategoryScheme",
"CATEGORYNAME":"CategoryName",
"CATEGORY":"Category",
"SECONDARYCATEGORY": "SecondaryCategory",
"SECONDARYCATEGORYNAME": "SecondaryCategoryName",
"TERTIARYCATEGORY": "TertiaryCategory",
"TERTIARYCATEGORYNAME":"TertiaryCategoryName",
"ACCOUNTAVERAGEWEIGHT":"AccountAverageWeight",
"ACCOUNTTOTALRETURN":"AccountTotalReturn",
"BENCHMARKAVERAGEWEIGHT":"BenchmarkAverageWeight",
"BENCHMARKTOTALRETURN":"BenchmarkTotalReturn",
"VARIATIONAVERAGEWEIGHT":"VariationAverageWeight",
"VARIATIONTOTALRETURN":"VariationTotalReturn",
"ALLOCATIONEFFECT":"AllocationEffect",
"SELECTIONANDINTERACTIONEFFECT":"SelectionAndInteractionEffect",
"TOTALEFFECT":"TotalEffect",
"WITHCASH": "WithCash"
}
blueprint = {
"PortfolioCode":"",
"CategoryScheme":"",
"Category":"",
"CategoryName":"",
"SecondaryCategory":"",
"SecondaryCategoryName":"",
"AccountAverageWeight":"",
"AccountTotalReturn":"",
"BenchmarkAverageWeight":"",
"BenchmarkTotalReturn":"",
"VariationAverageWeight":"",
"VariationTotalReturn":"",
"AllocationEffect":"",
"SelectionAndInteractionEffect":"",
"TotalEffect":"",
"WithCash":""
}
# CategoryLevelData_df = CategoryLevelData_df.replace(["NULL", "null"], np.nan)
CategoryLevelData_df = CategoryLevelData_df.fillna("")
CategoryLevelData_df = CategoryLevelData_df.rename(columns=transform_column_map)
if WithCash == 1 and ShowCash != 1:
CategoryLevelData_df = CategoryLevelData_df.loc[CategoryLevelData_df['CategoryName'] != '[Cash]'].reset_index()
else:
CategoryLevelData_df=CategoryLevelData_df
for j in range(CategoryLevelData_df.shape[0]):
temp_dict = CategoryLevelData_df.loc[j].to_dict()
keylist = list(temp_dict.keys())
for element in keylist:
if (element not in blueprint):
del temp_dict[element]
keylist = list(blueprint.keys())
for element in keylist:
if (element not in temp_dict.keys()):
temp_dict[element] = blueprint[element]
final_result_list.append(temp_dict)
response["data"] = final_result_list
Meta data:
Object Columns:
Request Parameter:
Dependency:
Dependency Block- DEMO_AccountbyPurpose setup:
Definition of DEMO_AccountbyPurpose Block:
AccountCode = ""
Purpose = ""
ShowComposite = ""
if params.__len__() > 0:
AccountCode = params["AccountCode"]
Purpose = params["Purpose"]
ShowComposite = params["ShowComposite"]
shift_account_data = read("DEMO_GetRepCompositeAccountbyProduct", {"AccountCode": AccountCode})["data"]
# print(shift_account_data) # List of Dictionories
display_account = ""
if (Purpose.lower() == "clients") and (ShowComposite.lower() == "no"):
display_account = AccountCode
else:
if len(shift_account_data) == 1:
product_data = shift_account_data[0]
# print(product_data.type)
if (Purpose.lower() == "clients") and (ShowComposite.lower() == "yes"):
display_account = product_data['PERFORMANCEACCOUNT']
elif (Purpose.lower() == "sales") and (ShowComposite.lower() == "no"):
display_account = product_data['PERFORMANCEACCOUNT']
elif (Purpose.lower() == "sales") and (ShowComposite.lower() == "yes"):
display_account = product_data['PERFORMANCEACCOUNT']
else:
display_account = AccountCode
response["values"] = display_account
Dependency of DEMO_AccountbyPurpose
Definition of DEMO_GetRepCompositeAccountbyProduct
Setting up Unit Tests for Attribution
Create sample data json files which we are going to access through the attribution transformation block. 1.1 Execute ‘DEMO_AccountbyPurpose’ and get value output and save in json format with same block name.
Execute “DEMO_GetDefaultCurruncyCode” interface block and save output data in json format with same block name.
Execute “DEMO_SnowFlakeAttributionDetailsExtract” main attribution data extraction block and save in same location in json format with same block name.
2.0 Use below sample unite test code and try to get correct output.
import json
import pandas as pd
import numpy as np
import os
def read_data(name, parameters={}):
file_name = name + '.json'
base_path2 = os.path.abspath(os.path.dirname(__file__))
base_path = os.path.dirname(base_path2)
file_path = os.path.join(base_path, 'Data\\' + file_name)
# if name == "DEMO_SnowFlakeAttributionDetailsExtract":
arr = {}
with open(file_path) as input_file:
json_data_file = json.load(input_file)
df = pd.DataFrame(json_data_file)
arr['data'] = json_data_file
filtered = []
for item in arr["data"]:
filtered.append(item)
if name== "DEMO_AccountbyPurpose":
return {"values": filtered}
elif name == "DEMO_GetDefaultCurruncyCode":
return {"values": filtered}
else:
return {"data": filtered}
# else:
# raise Exception("Unsupported block name")
def exec_script(response, read, params):
# Parameters for retreiving Attribution data
AccountCode=""
AsofDate=""
CurrencyCode=""
ShowComposite=""
Purpose=""
WithCash=""
CategoryScheme=""
Category=""
Period= ""
ShowCash=""
if(params.__len__()>0):
AccountCode = params["AccountCode"]
AsofDate = params["AsofDate"]
if("CurrencyCode" in params.keys()):
CurrencyCode = params["CurrencyCode"]
if("ShowComposite" in params.keys()):
ShowComposite = params["ShowComposite"]
if("Purpose" in params.keys()):
Purpose = params["Purpose"]
if("CategoryScheme" in params.keys()):
CategoryScheme = params["CategoryScheme"]
if("Category" in params.keys()):
Category = params["Category"]
if("WithCash" in params.keys()):
WithCash = params["WithCash"]
if("ShowCash" in params.keys()):
ShowCash = params["ShowCash"]
if("Period" in params.keys()):
Period = params["Period"]
#Getting accountcode
AccountCode = read("DEMO_AccountbyPurpose",{"AccountCode":AccountCode,"Purpose":Purpose,"ShowComposite":ShowComposite})["values"]
AccountCode=AccountCode
if CurrencyCode.lower() == "default":
setCurrency= read("DEMO_GetDefaultCurruncyCode", {"AccountCode":AccountCode})["values"]
else:
setCurrency=CurrencyCode
# Read data block to retrieve Attribution data
detailsdata_df = pd.DataFrame(read("DEMO_SnowFlakeAttributionDetailsExtract",{"AccountCode":AccountCode,"AsofDate":AsofDate,"CategoryScheme":CategoryScheme,"Category":Category,"WithCash":WithCash,"CurrencyCode":setCurrency,"Period":Period})["data"])
totaldata_df= pd.DataFrame(read("DEMO_SnowFlakeAttributionDetailsExtract",{"AccountCode":AccountCode,"AsofDate":AsofDate,"CategoryScheme":CategoryScheme,"Category":"Total","WithCash":WithCash,"CurrencyCode":setCurrency,"Period":Period})["data"])
data_df = pd.concat([detailsdata_df, totaldata_df], axis=0, ignore_index=True)
CategoryLevelData_df = data_df.loc[data_df['SECONDARYCATEGORY'] .isnull()].reset_index(drop=True)
final_result_list = []
transform_column_map = {
"PORTFOLIOCODE":"PortfolioCode",
"CATEGORYSCHEME":"CategoryScheme",
"CATEGORYNAME":"CategoryName",
"CATEGORY":"Category",
"SECONDARYCATEGORY": "SecondaryCategory",
"SECONDARYCATEGORYNAME": "SecondaryCategoryName",
"TERTIARYCATEGORY": "TertiaryCategory",
"TERTIARYCATEGORYNAME":"TertiaryCategoryName",
"ACCOUNTAVERAGEWEIGHT":"AccountAverageWeight",
"ACCOUNTTOTALRETURN":"AccountTotalReturn",
"BENCHMARKAVERAGEWEIGHT":"BenchmarkAverageWeight",
"BENCHMARKTOTALRETURN":"BenchmarkTotalReturn",
"VARIATIONAVERAGEWEIGHT":"VariationAverageWeight",
"VARIATIONTOTALRETURN":"VariationTotalReturn",
"ALLOCATIONEFFECT":"AllocationEffect",
"SELECTIONANDINTERACTIONEFFECT":"SelectionAndInteractionEffect",
"TOTALEFFECT":"TotalEffect",
"WITHCASH": "WithCash"
}
blueprint = {
"PortfolioCode":"",
"CategoryScheme":"",
"Category":"",
"CategoryName":"",
"SecondaryCategory":"",
"SecondaryCategoryName":"",
"AccountAverageWeight":"",
"AccountTotalReturn":"",
"BenchmarkAverageWeight":"",
"BenchmarkTotalReturn":"",
"VariationAverageWeight":"",
"VariationTotalReturn":"",
"AllocationEffect":"",
"SelectionAndInteractionEffect":"",
"TotalEffect":"",
"WithCash":""
}
# CategoryLevelData_df = CategoryLevelData_df.replace(["NULL", "null"], np.nan)
CategoryLevelData_df = CategoryLevelData_df.fillna("")
CategoryLevelData_df = CategoryLevelData_df.rename(columns=transform_column_map)
if WithCash == 1 and ShowCash != 1:
CategoryLevelData_df = CategoryLevelData_df.loc[CategoryLevelData_df['CategoryName'] != '[Cash]'].reset_index()
else:
CategoryLevelData_df=CategoryLevelData_df
for j in range(CategoryLevelData_df.shape[0]):
temp_dict = CategoryLevelData_df.loc[j].to_dict()
keylist = list(temp_dict.keys())
for element in keylist:
if (element not in blueprint):
del temp_dict[element]
keylist = list(blueprint.keys())
for element in keylist:
if (element not in temp_dict.keys()):
temp_dict[element] = blueprint[element]
final_result_list.append(temp_dict)
# for j in range(totaldata_df.shape[0]):
# temp_dict = totaldata_df.loc[j].to_dict()
# keylist = list(temp_dict.keys())
# for element in keylist:
# if (element not in blueprint):
# del temp_dict[element]
# keylist = list(blueprint.keys())
# for element in keylist:
# if (element not in temp_dict.keys()):
# temp_dict[element] = blueprint[element]
# final_result_list.append(temp_dict)
response["data"] = final_result_list
# print output into data file
output_filename = 'data2.json'
with open(output_filename, 'w') as outfile:
json.dump(final_result_list, outfile)
def test_Attribution_data():
parameters = {"AccountCode": "bbdlcvmodel", "AsofDate": "2023-04-30", "CurrencyCode": "USD", "CategoryScheme":"GICS Sector","Category":"Sector", "Period":"MTD", "WithCash": 1, "ShowCash": 0, "ShowComposite": "no" }
response = {}
exec_script(response, read_data, parameters)
assert "data" in response
data = response["data"]
# print(data)
# assert len(data) == 22
test_Attribution_data()
Creating the Data Attribution Object
Here we can use the Data Block ‘DEMO_AttributionCategoryDetails to create the Data object ‘DEMO_Attribution With Total’.
Click on ‘Create New’ under ‘Data Objects’ tab from Developer Center
Select the options as shown below and hit ‘Create’
Give an appropriate name.
Set your Data Block
Set the Data Settings one by one
Set the Column name one by one as needed.
Go to Sorting and can set the sorting option as shown below.
Set the Row Type as shown below.
Authoring Smart page (Attribution - Table)
Insert a native PowerPoint table as shown below (8 Columns by 3 Rows table structure)
Open Fabrication panel as shown below.
Set the Columns, Rows and Data Settings as shown below.