Analyse AWS billing report

aws-billing.py

#!/usr/bin/env python3

import json
import glob
import argparse

import pandas as pd


ITEM_TYPE = "lineItem/LineItemType"
ITEM_PRODUCT_CODE = "lineItem/ProductCode"
ITEM_USAGE_TYPE = "lineItem/UsageType"
ITEM_UNBLENDED_COST = "lineItem/UnblendedCost"
ITEM_EFFECTIVE_COST = "reservation/EffectiveCost"


def read_manifest(name):
    manifest_file = f"{name}-Manifest.json"
    print(f"    Reading manifest {manifest_file} ...")
    metadata = json.load(open(manifest_file))
    columns = metadata["columns"]
    headers = []
    dtypes = {}
    parse_dates = []
    for column in columns:
        cname = "{}/{}".format(column["category"], column["name"])
        headers.append(cname)
        if column["type"] == "DateTime":
            dtypes[cname] = "str"
            parse_dates.append(cname)
        elif column["type"] in ["BigDecimal", "OptionalBigDecimal"]:
            dtypes[cname] = "float"
        else:
            dtypes[cname] = "str"
    return headers, dtypes, parse_dates


def read_report(name):
    headers, dtypes, parse_dates = read_manifest(name)
    report_files = glob.glob(f"{name}-*.csv.gz")
    dfs = []
    for report_file in report_files:
        print(f"    Reading report {report_file} ...")
        df = pd.read_csv(
            report_file,
            compression="gzip",
            names=headers,
            dtype=dtypes,
            parse_dates=parse_dates,
            skiprows=1,
            low_memory=False,
            infer_datetime_format=True,
        )
        dfs.append(df)
    data = pd.concat(dfs, axis=0, ignore_index=True)
    return data


def analyze_report(data):
    print("--> Analyzing billing report ...")
    print("-" * 40)
    tax = data[data[ITEM_TYPE] == "Tax"]
    tax_cost = tax[ITEM_UNBLENDED_COST].sum()
    print("Tax: ${:.2f}".format(tax_cost))
    if tax_cost > 0.01:
        product_cost = tax.groupby(ITEM_PRODUCT_CODE)[ITEM_UNBLENDED_COST].sum()
        for product, cost in product_cost.items():
            if tax < 0.01:
                continue
            print(f"  {product}: ${cost:.2f}")
    print("-" * 40)
    usage = data[data[ITEM_TYPE] == "Usage"]
    usage_cost = usage[ITEM_UNBLENDED_COST].sum()
    print("Usage: ${:.2f}".format(usage_cost))
    if usage_cost > 0.01:
        product_cost = usage.groupby(ITEM_PRODUCT_CODE)[ITEM_UNBLENDED_COST].sum()
        for product, cost in product_cost.items():
            if cost < 0.01:
                continue
            print(f"  {product}: ${cost:.2f}")
            detail = (
                usage[usage[ITEM_PRODUCT_CODE] == product]
                .groupby(ITEM_USAGE_TYPE)[ITEM_UNBLENDED_COST]
                .sum()
            )
            for t, s in detail.items():
                if s < 0.01:
                    continue
                print(f"    {t}: ${s:.2f}")
    print("-" * 40)
    discount = data[data[ITEM_TYPE] == "DiscountedUsage"]
    discount_cost = discount[ITEM_EFFECTIVE_COST].sum()
    print("Discounted usage (effective cost): ${:.2f}".format(discount_cost))
    if discount_cost > 0.01:
        product_cost = discount.groupby(ITEM_PRODUCT_CODE)[ITEM_EFFECTIVE_COST].sum()
        for product, cost in product_cost.items():
            if cost < 0.01:
                continue
            print(f"  {product}: ${cost:.2f}")
            detail = (
                discount[discount[ITEM_PRODUCT_CODE] == product]
                .groupby(ITEM_USAGE_TYPE)[ITEM_EFFECTIVE_COST]
                .sum()
            )
            for t, s in detail.items():
                if s < 0.01:
                    continue
                print(f"    {t}: ${s:.2f}")
    print("=" * 40)
    print("Total: ${:.2f}".format(tax_cost + usage_cost + discount_cost))


def main():
    parser = argparse.ArgumentParser(
        prog="AWSBilling", description="Analyse AWS billing report"
    )
    parser.add_argument("name", help="Name of the billing report to analyse", type=str)
    args = parser.parse_args()

    print(f"--> Reading billing report {args.name} ...")
    data = read_report(args.name)
    analyze_report(data)


if __name__ == "__main__":
    main()