KPLER_USERNAME=user@example.com
KPLER_PASSWORD=your_password

cp .claude/skills/kpler/scripts/kpler_client.py scripts/

uv add httpx pyjwt python-dotenv

import asyncio
from kpler_client import KplerClient

async def main():
    async with KplerClient() as client:
        # Search for a player
        results = await client.search("shell", categories=["PLAYER"])
        shell_id = results["players"][0]["entity"]["id"]

        # Query their trades
        trades = await client.query_trades(players=[shell_id], size=20)
        print(f"Found {len(trades['result']['trades'])} trades")

asyncio.run(main())

client = KplerClient()
await client.login("user@example.com", "password")  # Manual login
client.is_authenticated()  # Check status
await client.logout()  # Clear tokens

results = await client.search(
    text="gazprom",
    categories=["PLAYER", "INSTALLATION"],  # Optional filter
    commodity_types=["lng"],  # Optional: lng, oil, lpg, dry
)

# Results grouped by type
for player in results.get("players", []):
    print(f"{player['entity']['name']} (ID: {player['entity']['id']})")

trades = await client.query_trades(
    # Pagination
    from_=0,
    size=100,

    # Filters (use IDs from search - strings work, converted to ints)
    locations=[1234],  # Zone/installation IDs
    products=[5678],   # Product IDs
    players=[3836],    # Company IDs
    vessels=[9012],    # Vessel IDs

    # Trade status
    statuses=["completed"],  # ongoing, completed, cancelled
    trade_types=["import", "export"],

    # Options
    with_forecasted=False,
)

for trade in trades["data"]:
    origin = trade.get("portCallOrigin", {}).get("zone", {}).get("name", "Unknown")
    dest = trade.get("portCallDestination", {}).get("zone", {}).get("name", "Unknown")
    print(f"{origin} → {dest}")

flows = await client.query_flows(
    # Required
    direction="export",  # export, import
    granularity="months",  # years, months, weeks, days

    # Date range
    start_date="2024-01-01",
    end_date="2024-12-31",

    # Filters
    locations=[1234],
    products=[5678],
    players=[3836],

    # Split results by dimension
    split_on="countries",  # countries, ports, products, vessels, buyers, sellers

    # Options
    cumulative=False,
    forecasted=False,
    intra=False,  # Include intra-region flows
)

# Response format: series by date with split values
for entry in flows["series"]:
    year = entry["date"]
    for dataset in entry.get("datasets", []):
        for split in dataset.get("splitValues", []):
            vol = split["values"]["volume"]
            print(f"{year}: {split['name']} = {vol/1e6:.1f} Mt")

positions = await client.get_vessel_positions(
    vessel_id=12345,
    start_date="2024-01-01T00:00:00Z",  # ISO 8601
    end_date="2024-01-31T23:59:59Z",
    limit=1000,
)

for pos in positions:
    print(f"{pos['timestamp']}: {pos['lat']}, {pos['lon']}")

fleet = await client.get_player_fleet(player_id=3836)

print(f"Company: {fleet['name']}")
print(f"Vessels owned: {len(fleet.get('ownedVessels', []))}")
print(f"Subsidiaries: {len(fleet.get('subsidiaries', []))}")

contracts = await client.query_contracts(
    types=["SPA", "LTA"],  # Tender, SPA, LTA, TUA
    players=[3836],
    from_=0,
    size=50,
)

# scripts/fetch_kpler.py
import asyncio
import duckdb
from kpler_client import KplerClient
from dotenv import load_dotenv

load_dotenv()

async def fetch_data():
    async with KplerClient() as client:
        # Search for entities
        russia = await client.search("russia", categories=["ZONE"])
        russia_id = russia["zones"][0]["entity"]["id"]

        # Get export flows
        flows = await client.query_flows(
            direction="export",
            locations=[russia_id],
            granularity="months",
            start_date="2020-01-01",
            end_date="2024-12-31",
            split_on="countries",
        )

        return flows["result"]["series"]

def build_database(data):
    con = duckdb.connect("data/data.duckdb")
    con.execute("""
        CREATE OR REPLACE TABLE flows (
            date DATE,
            destination VARCHAR,
            volume_kt DOUBLE
        )
    """)

    for series in data:
        for point in series.get("data", []):
            con.execute(
                "INSERT INTO flows VALUES (?, ?, ?)",
                [point["date"], series["name"], point["value"]]
            )

    con.close()

if __name__ == "__main__":
    data = asyncio.run(fetch_data())
    build_database(data)

data:
	uv run python scripts/fetch_kpler.py

import asyncio

for player_id in player_ids:
    data = await client.query_trades(players=[player_id])
    await asyncio.sleep(0.5)  # Rate limit protection