1#!/usr/bin/env python3
2"""
3ElastiCache Price Calculator
4
5Fetches live pricing from AWS and estimates costs for specific configurations.
6
7Supports:
8- Serverless (Valkey, Redis OSS, Memcached)
9- Node-based on-demand and reserved instances
10- Multi-node clusters with replicas
11- Extended Support surcharge for EOL Redis versions
12
13Usage:
14  # Serverless estimate
15  python3 price_calculator.py --engine valkey --mode serverless --data-gb 5 --ecpu-millions 200
16
17

price_calculator.py on GitHub

# Node-based with replicas

python3 price_calculator.py --engine valkey --mode node --node-type cache.r7g.large --nodes 3

# Node-based with specific RI term

python3 price_calculator.py --mode node --node-type cache.r7g.large --nodes 2 --ri-term 1yr_no_upfront

# Show all reserved options for a node type

python3 price_calculator.py --mode node --node-type cache.r7g.large --nodes 2 --show-ri-options

# Specific region

python3 price_calculator.py --engine valkey --region eu-west-1 --mode serverless --data-gb 10

# Extended Support surcharge for EOL Redis versions

python3 price_calculator.py --engine redis --extended-support --node-type cache.r7g.large --nodes 6

# Interactive mode

python3 price_calculator.py --interactive

"""

import argparse

import json

import os

import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from pricing import PricingLoader

# Detect region from environment with us-east-1 as fallback

REGION = os.environ.get("AWS_REGION") or os.environ.get("AWS_DEFAULT_REGION") or "us-east-1"

HOURS_PER_MONTH = 730

PRICING_DISCLAIMER = (

" Estimates only. Verify current pricing at https://aws.amazon.com/elasticache/pricing/"

)

def estimate_serverless(loader, engine, region, data_gb, ecpu_millions_per_month, avg_payload_kb=1.0):

storage_rate = loader.get_serverless_storage_rate(region, engine)

ecpu_rate = loader.get_serverless_ecpu_rate(region, engine)

min_gb = 0.1 if engine == "valkey" else 1.0

billing_gb = max(data_gb, min_gb)

storage = billing_gb * storage_rate * HOURS_PER_MONTH

scaled_ecpus = ecpu_millions_per_month * max(1.0, avg_payload_kb)

compute = scaled_ecpus * 1_000_000 * ecpu_rate

result = {

"model": "serverless",

"engine": engine,

"region": region,

"data_gb": data_gb,

"ecpu_millions": ecpu_millions_per_month,

"storage_monthly": round(storage, 2),

"compute_monthly": round(compute, 2),

"total_monthly": round(storage + compute, 2),

}

if avg_payload_kb != 1.0:

result["avg_payload_kb"] = avg_payload_kb

result["ecpu_millions_scaled"] = round(scaled_ecpus, 1)

return result

def estimate_node_based(loader, engine, region, node_type, node_count, ri_term=None):

try:

hourly = loader.get_node_hourly_rate(region, node_type, engine)

except ValueError as e:

return {"error": str(e)}

monthly_per_node = hourly * HOURS_PER_MONTH

total = monthly_per_node * node_count

mem = loader.get_node_memory_gb(node_type)

result = {

"model": "node-based (on-demand)",

"engine": engine,

"region": region,

"node_type": node_type,

"node_count": node_count,

"memory_per_node_gb": mem if mem > 0 else "unknown",

"total_memory_gb": round(mem * node_count, 1) if mem > 0 else "unknown",

"per_node_monthly": round(monthly_per_node, 2),

"total_monthly": round(total, 2),

}

# Try real reserved pricing first, fall back to hardcoded discount

if ri_term:

ri_map = {

"1yr_no_upfront": ("1yr", "No Upfront"),

"1yr_partial_upfront": ("1yr", "Partial Upfront"),

"1yr_all_upfront": ("1yr", "All Upfront"),

"3yr_no_upfront": ("3yr", "No Upfront"),

"3yr_partial_upfront": ("3yr", "Partial Upfront"),

"3yr_all_upfront": ("3yr", "All Upfront"),

}

if ri_term in ri_map:

lease, purchase = ri_map[ri_term]

try:

eff_monthly = loader.get_reserved_effective_monthly(

region, node_type, engine, lease, purchase

)

ri_total = eff_monthly * node_count

discount_pct = round((1 - ri_total / total) * 100)

result["reserved"] = {

"term": ri_term,

"discount_pct": discount_pct,

"per_node_monthly": eff_monthly,

"total_monthly": round(ri_total, 2),

"annual_savings": round((total - ri_total) * 12, 2),

"source": "live",

}

except ValueError:

pass # no reserved pricing available for this combo

return result

def format_ri_options(loader, engine, region, node_type, node_count):

"""Show all reserved pricing options for a specific node configuration."""

options = loader.list_reserved_options(region, node_type, engine)

hourly = loader.get_node_hourly_rate(region, node_type, engine)

od_monthly = hourly * HOURS_PER_MONTH * node_count

lines = []

lines.append("=" * 68)

lines.append("Reserved Pricing Options (live pricing)")

lines.append("=" * 68)

lines.append("")

lines.append(" Node type: {} x {} ({} engine)".format(node_type, node_count, engine))

lines.append(" On-demand: ${:,.2f}/month".format(od_monthly))

lines.append("")

if not options:

lines.append(" No reserved pricing options found for this configuration.")

lines.append("")

lines.append(PRICING_DISCLAIMER)

lines.append("")

return "\n".join(lines)

lines.append(" {:<10} {:<18} {:>12} {:>10} {:>10}".format(

"Term", "Purchase Option", "Monthly", "Upfront", "Savings"))

lines.append(" {} {} {} {} {}".format("-" * 10, "-" * 18, "-" * 12, "-" * 10, "-" * 10))

for opt in options:

eff_total = opt["effective_monthly"] * node_count

savings_pct = round((1 - eff_total / od_monthly) * 100)

upfront_total = opt["upfront"] * node_count

lines.append(" {:<10} {:<18} ${:>9,.2f} ${:>8,.0f} {:>8}%".format(

opt["lease_length"], opt["purchase_option"],

eff_total, upfront_total, savings_pct))

lines.append("")

lines.append(PRICING_DISCLAIMER)

lines.append("")

return "\n".join(lines)

def estimate_extended_support(loader, engine, region, node_type, node_count):

"""Estimate Extended Support surcharge for EOL Redis versions."""

results = {"engine": engine, "region": region, "node_type": node_type, "node_count": node_count, "years": []}

if not loader.has_extended_support_pricing():

results["note"] = "Extended Support pricing not available in bulk pricing API. Check the ElastiCache pricing page."

return results

for year in ["1", "2", "3"]:

try:

hourly = loader.get_extended_support_rate(region, node_type, engine, year)

monthly = hourly * HOURS_PER_MONTH * node_count

results["years"].append({

"year": int(year),

"per_node_hourly": hourly,

"total_monthly": round(monthly, 2),

"total_annual": round(monthly * 12, 2),

})

except ValueError:

pass

if results["years"]:

try:

base_hourly = loader.get_node_hourly_rate(region, node_type, engine)

base_monthly = base_hourly * HOURS_PER_MONTH * node_count

results["base_monthly"] = round(base_monthly, 2)

total_yr1 = base_monthly + results["years"][0]["total_monthly"]

results["total_with_surcharge_monthly"] = round(total_yr1, 2)

except ValueError:

pass

results["recommendation"] = (

"Migrating to Valkey eliminates Extended Support charges. "

"Savings vs other engines: 20% lower on node-based, 33% lower on serverless."

)

return results

def format_extended_support_report(results):

"""Generate a human-readable Extended Support cost report."""

lines = []

lines.append("=" * 60)

lines.append("Extended Support Cost Estimate (live pricing)")

lines.append("=" * 60)

lines.append("")

lines.append(" Engine: {}".format(results["engine"]))

lines.append(" Region: {}".format(results["region"]))

lines.append(" Node type: {}".format(results.get("node_type", "unknown")))

lines.append(" Nodes: {}".format(results["node_count"]))

lines.append("")

if not results["years"]:

lines.append(" {}".format(results.get("note", "No pricing data available.")))

lines.append("")

return "\n".join(lines)

lines.append(" {:<18} {:>16} {:>16}".format("Year After EOL", "Monthly Cost", "Annual Cost"))

lines.append(" {} {} {}".format("-" * 18, "-" * 16, "-" * 16))

for y in results["years"]:

lines.append(" Year {:<13} ${:>13,.2f} ${:>13,.2f}".format(

y["year"], y["total_monthly"], y["total_annual"]

))

if results.get("base_monthly"):

lines.append("")

lines.append(" Base node cost: ${:>13,.2f}/month".format(results["base_monthly"]))

lines.append(" Total (base + Yr1): ${:>13,.2f}/month".format(results["total_with_surcharge_monthly"]))

lines.append("")

if "recommendation" in results:

lines.append(" {}".format(results["recommendation"]))

lines.append("")

lines.append(PRICING_DISCLAIMER)

lines.append("")

return "\n".join(lines)

def estimate_ecpu_from_ops(ops_per_sec, avg_ecpu_per_op=1.0):

"""Convert operations per second to monthly ECPUs (millions).

Default: 1 ECPU per operation (assumes simple GET/SET with payload under 1 KB).

For larger payloads, use --avg-payload-kb which scales ECPUs linearly.

For complex commands (SORT, ZADD, etc.), pass a higher avg_ecpu_per_op.

"""

ecpu_per_month = ops_per_sec * avg_ecpu_per_op * 3600 * HOURS_PER_MONTH

return round(ecpu_per_month / 1_000_000, 1)

def interactive(loader, region):

"""Interactive interview-style cost estimation."""

print("=== ElastiCache Price Calculator (live pricing) ===\n")

engine = input("Engine [valkey/redis/memcached] (default: valkey): ").strip().lower() or "valkey"

data_gb = float(input("Estimated data size in GB (default: 1): ").strip() or "1")

mode = input("Deployment [serverless/node] (default: serverless): ").strip().lower() or "serverless"

if mode == "serverless":

ops = input("Estimated operations per second (default: 100): ").strip() or "100"

ops_per_sec = float(ops)

ecpu_millions = estimate_ecpu_from_ops(ops_per_sec)

print(" -> Estimated {}M ECPUs/month".format(ecpu_millions))

result = estimate_serverless(loader, engine, region, data_gb, ecpu_millions)

print("\n{}".format(json.dumps(result, indent=2)))

elif mode == "node":

node_type = input("Node type (default: cache.r7g.large): ").strip() or "cache.r7g.large"

nodes = int(input("Number of nodes (default: 2): ").strip() or "2")

result = estimate_node_based(loader, engine, region, node_type, nodes, ri_term="1yr_no_upfront")

print("\n{}".format(json.dumps(result, indent=2)))

show_ri = input("\nShow all reserved options? [y/n] (default: n): ").strip().lower()

if show_ri == "y":

print("\n{}".format(format_ri_options(loader, engine, region, node_type, nodes)))

if __name__ == "__main__":

parser = argparse.ArgumentParser(description="ElastiCache Price Calculator (live pricing)")

parser.add_argument("--interactive", action="store_true", help="Interactive mode")

parser.add_argument("--engine", choices=["valkey", "redis", "memcached"], default="valkey")

parser.add_argument("--mode", choices=["serverless", "node"], default="serverless")

parser.add_argument("--region", default=REGION, help="AWS region (default: from env or us-east-1)")

parser.add_argument("--data-gb", type=float, default=1.0, help="Data stored in GB")

parser.add_argument("--ecpu-millions", type=float, default=None, help="ECPUs per month (millions)")

parser.add_argument("--ops-per-sec", type=float, default=None, help="Operations per second (auto-converts to ECPUs)")

parser.add_argument("--node-type", default="cache.r7g.large", help="Node type for node-based")

parser.add_argument("--nodes", type=int, default=2, help="Node count for node-based")

parser.add_argument("--ri-term", default=None,

choices=["1yr_no_upfront", "1yr_partial_upfront", "1yr_all_upfront",

"3yr_no_upfront", "3yr_partial_upfront", "3yr_all_upfront"],

help="Reserved instance term")

parser.add_argument("--show-ri-options", action="store_true",

help="Show all reserved pricing options for the node type")

parser.add_argument("--extended-support", action="store_true",

help="Show Extended Support surcharge for EOL Redis versions")

parser.add_argument("--avg-payload-kb", type=float, default=1.0,

help="Average payload size in KB (default: 1.0). ECPUs scale linearly with payload.")

parser.add_argument("--pricing-csv", default=None, help="Optional local pricing CSV (overrides live fetch)")

args = parser.parse_args()

loader = PricingLoader(args.pricing_csv)

if args.interactive:

interactive(loader, args.region)

sys.exit(0)

# Auto-convert ops/sec to ECPUs if provided

ecpu_millions = args.ecpu_millions

if ecpu_millions is None and args.ops_per_sec:

ecpu_millions = estimate_ecpu_from_ops(args.ops_per_sec)

elif ecpu_millions is None:

ecpu_millions = 100.0

if args.extended_support:

result = estimate_extended_support(loader, args.engine, args.region, args.node_type, args.nodes)

print(format_extended_support_report(result))

elif args.show_ri_options:

print(format_ri_options(loader, args.engine, args.region, args.node_type, args.nodes))

elif args.mode == "serverless":

result = estimate_serverless(loader, args.engine, args.region, args.data_gb, ecpu_millions, args.avg_payload_kb)

result["disclaimer"] = "Estimates only. Verify at https://aws.amazon.com/elasticache/pricing/"

print(json.dumps(result, indent=2))

elif args.mode == "node":

result = estimate_node_based(loader, args.engine, args.region, args.node_type, args.nodes, ri_term=args.ri_term)

result["disclaimer"] = "Estimates only. Verify at https://aws.amazon.com/elasticache/pricing/"

print(json.dumps(result, indent=2))