-
Notifications
You must be signed in to change notification settings - Fork 1
/
income_reporter.py
130 lines (105 loc) · 4.47 KB
/
income_reporter.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
# System libraries
from datetime import datetime
import sys
# Internal library imports
from utility.api_calls import get_coin_balance_history
from utility.input_checks import check_blockscout_api, check_coinmarketcap_api
from utility.input_checks import is_valid_eth_address, is_valid_year, check_file
from utility.terminal_outputs import printLine, printHead, printFoot, printIntro
from utility.price_calculation import create_daily_data_with_prices, calculate_total_income
from utility.pdf_generation import csv_to_pdf
from utility.csv_exports import export_to_csv
# Internal config data
from config import BLOCKSCOUT_API_URL, COINMARKETCAP_HEADERS, COINMARKETCAP_API_URL
from config import ETH1_ADDRESS, YEAR, COIN_NAME, FIAT_CURRENCY
def generate_income_report():
# Generates the income report.
# Start terminal outputs
printHead()
printIntro()
""""
Check if config data is valid.
- Blockscout API must be reachable
- CoinMarketCap API must be accessible with API KEY
- Address must be valid ETH1 address
- Year must be a valid number
"""
if not (check_blockscout_api(BLOCKSCOUT_API_URL) and
check_coinmarketcap_api(COINMARKETCAP_API_URL, COINMARKETCAP_HEADERS) and
is_valid_eth_address(ETH1_ADDRESS) and
is_valid_year(YEAR)):
printLine()
printLine("❌ Input validation failed. Exiting program.", True)
printFoot()
return
printFoot()
"""
Check for existing CSV and PDF files
in the current folder and with the same name
"""
file_name = f"income_report_{YEAR}_{ETH1_ADDRESS}"
if not check_file(file_name):
sys.exit("++ Operation aborted. File was not overwritten.\n")
start_time = datetime.now()
print(f"Starting income report at {start_time.strftime('%Y-%m-%d %H:%M')}")
# Fetch income + withdrawal data from Blockscout
printHead()
daily_deltas, miner_count, withdrawal_count = get_coin_balance_history()
validator_earnings = miner_count + withdrawal_count
"""
Get price history from CoinMarketCap
for every day with income, e.g. positive deltas
"""
daily_data = create_daily_data_with_prices(daily_deltas)
"""
Export income data into a CSV file including
- dates with their deltas
- daily price and income
"""
csv_file_name = f"{file_name}.csv"
export_to_csv(csv_file_name, daily_data, ['Date', 'Received ' + COIN_NAME, 'Former ' + COIN_NAME + ' Price', 'Income in ' + FIAT_CURRENCY])
printLine()
"""
Generate report data metrics:
- Calculate yearly income in FIAT and crypto currency
- Measure failure tolerance of days without price
- Show total withdrawal listings and miner records
"""
total_income, total_coins, missing_data_count = calculate_total_income(daily_data)
total_rows = len(daily_data)
total_coins_formatted = f"{total_coins:.8f}"
printLine(f"⏩ The address received a total of {validator_earnings} validator payments from:", True)
printLine(f"⏩ {withdrawal_count} withdrawal listings and {miner_count} miner records.", True)
printLine()
printLine(f"⏩ Received {total_coins_formatted} {COIN_NAME} in {YEAR} worth", True)
printLine(f"⏩ {total_income} {FIAT_CURRENCY} in price-adjusted income.", True)
printLine()
printLine(f"🔎 {missing_data_count} of {total_rows} days with income are missing price data", True)
printLine()
"""
Generate PDF validator report including:
- Front page with address, description, and links
- Table showing monthly and yearly income
- Detailed pages for every month, showing daily income
"""
pdf_file_name = f"{file_name}.pdf"
csv_to_pdf(csv_file_name, pdf_file_name, miner_count, withdrawal_count)
end_time = datetime.now()
# Calculate the duration
duration = end_time - start_time
# Format the duration as hours, minutes, and seconds
duration_seconds = duration.total_seconds()
hours, minutes = divmod(duration_seconds // 60, 60)
# End terminal outputs
printLine()
printLine("🏁 Income report finished successfully", True)
printFoot()
print(f"Stopping income report at {end_time.strftime('%Y-%m-%d %H:%M')} after {int(hours):02}:{int(minutes):02}h \n\n")
# Execute report when script is called
if __name__ == '__main__':
try:
generate_income_report()
# Script gets exited
except KeyboardInterrupt:
print("\n\nProgram interrupted by user. Exiting gracefully. \n")
sys.exit(0)