How to Build a PEAQ DePIN Network Monitor with CoinMarketCap API

Build a PEAQ DePIN network monitor using the CoinMarketCap API, PEAQ token tracking, machine economy detection, Ethereum DEX liquidity checks, macro filters, Python examples, and production tips.

introduction

PEAQ is a blockchain for the machine economy.

Where Helium focuses on wireless coverage and Akash focuses on compute, PEAQ focuses on machine identity and orchestration — enabling connected devices, autonomous vehicles, robots, and IoT sensors to have identities, earn income, and participate in decentralized networks. PEAQ’s ecosystem includes electric vehicle charging networks, autonomous vehicle data marketplaces, and industrial IoT platforms.

PEAQ acts as a Polkadot umbrella chain with EVM compatibility, making it accessible to Ethereum developers while taking advantage of Polkadot’s shared security. The PEAQ token is used for transaction fees, governance, and to reward machine operators.

By early 2026, PEAQ has connected over 10 million devices and hosts multiple DePIN protocols including ELOOP (EV sharing), MapMetrics (navigation data), and others based on its device identity layer.

Unlike Cosmos-based DePIN tokens, PEAQ’s EVM compatibility means the token may have liquidity in EVM-based DEXs. In this guide, you will build Network monitoring PEAQ DePIN with CoinMarketCap APIwhere:

• The CoinMarketCap API powers the market signals engine
• PEAQ’s official API and on-chain contracts handle real machine activity, operator rewards, and network state

Illustration of architecture

The CoinMarketCap API strictly acts as an off-chain signaling layer to monitor PEAQ token prices, detect the narrative trend of the machine economy, and filter the overall system. It’s not a device identity monitor, a device activity oracle, or an operator reward feed.

The real device count, device activity, operator rewards and TVL protocol must be validated via the official PEAQ API or RPC of the PEAQ chain.

Project preparation

Import operating system
Import time
Import date
Import requests

CMC_API_KEY = os.getenv(“CMC_API_KEY”)
CMC_BASE_URL = “https://pro-api.coinmarketcap.com”

vertices = {
“accept”: “application/json”,
“X-CMC_PRO_API_KEY”: CMC_API_KEY, }

# Machine Economy Sector DePIN
DEPIN_ASSETS= [“PEAQ”, “IOTX”, “HNT”, “RENDER”]

# DePIN tags for local filtering
DEPIN_TAGS = {“depin”,”dotted ecosystem”,”peakecosystem”,”IoT”}

Step 1: Asset map

Map_assets definition(symbols=”PEAQ, IOTX, HNT, RENDER”):
url = f”{CMC_BASE_URL}/v1/cryptocurrency/map”
parameters = {“symbol”:symbols}
r = request.get(url, headers=HEADERS, params=params)
r.raise_for_status()
returns r.json()[“data”]

Definition of Resolve_peaq_id(map_data):
For assets in Map_data:
If asset.get(“symbol”) == “PEAQ” and “peaq” in (asset.get(“slug”) or “”).lower():
Return of assets[“id”]
Next return ((a[“id”] to get Map_data if a.get(“symbol”) == “PEAQ”), None)

Step 2: Fetch quotes

Uncheck fetch_quotes(ids):
url = f”{CMC_BASE_URL}/v3/cryptocurrency/quotes/latest”
Parameters = {“id”: “,”.join(str(i) for i in ids)}
r = request.get(url, headers=HEADERS, params=params)
r.raise_for_status()
returns r.json()[“data”]

native parse_quote(assets):
usd = next((q for q in assets.get(“quote”, []) if q.get(“symbol”) == “USD”), {})
returns {
“id”: assets.get(“id”),
“symbol”: assets.get(“symbol”),
“Price”: usd.get(“Price”),
“volume_24h”: usd.get(“volume_24h”),
“market_cap”: usd.get(“market_cap”),
“fdv”: usd.get(“full_diluted_market_cap”),
“pct_change_1h”: usd.get(“percent_change_1h”),
“pct_change_24h”: usd.get(“percent_change_24h”),
“pct_change_7d”: usd.get(“percent_change_7d”),
“tvl”: usd.get(“tvl”),
“tvl_ratio”: assets.get(“tvl_ratio”),
“num_market_pairs”: assets.get(“num_market_pairs”),
}

quotes = {str(a[“id”]): parse_quote(a) for raw_quotes}

Step 3: Record the peak

Definition of compute_peaq_score (quote):
Result = 0 pct_1h = quote.get(“pct_change_1h”) or 0
pct_24h = quote.get(“pct_change_24h”) or 0
pct_7d = quote.get(“pct_change_7d”) or 0

If pct_24h > 10: result += 30
Elif pct_24h > 5: result += 20
Elif pct_24h > 2: score += 10
elif pct_24h < -15: result -= 25
If pct_7d > 20: result += 20
elif pct_7d > 10: result += 10
If pct_1h > 2: result += 15
elif pct_1h > 0.5: score += 8

Volume = quote.get(“volume_24h”) or 0
If folder > 10_000_000: result += 20
elif folder > 2_000_000: result += 10

mcap = quote.get(“market_cap”) or 0
If mcap > 200_000_000: result += 15
Elif Makab > 50_000_000: result += 8

tvl_ratio = quote.get(“tvl_ratio”) or 0
If 0 < tvl_ratio < 1: score += 10

degree of return

Step 4: DEX Liquidity (if any)

PEAQ is compatible with EVM. Check liquidity on neighboring Ethereum or Polkadot-DEXs.

def fetch_peaq_pools(peaq_contract_address, platform=”ethereum”):
tries:
url = f”{CMC_BASE_URL}/v1/dex/token/pools”
parameters = {“address”: peaq_contract_address, “platform”: platform}
r = request.get(url, headers=HEADERS, params=params)
r.raise_for_status()
returns r.json()[“data”]
Except for the exception:
He comes back []

def get_best_pool(pools, min_liquidity=50_000):
Saleh = [p for p in pools if float(p.get(“liqUsd”) or 0) >= min_liquidity]
return max(valid, key=lambda p: float(p.get(“liqUsd”) or 0)) if valid otherwise nil

liqUsd is a string – it is always sent to float.

Step 5: Momentum Candle (if there is an EVM rally)

def fetch_candles(contract_address, platform=’ethereum’, interval=’1h”):
url = f”{CMC_BASE_URL}/v1/k-line/candles”
parameters = {“platform”:platform, “address”:contract_address, “interval”:timeinterval}
r = request.get(url, headers=HEADERS, params=params)
r.raise_for_status()
returns r.json()[“data”]

def parse_candle (c):
returns {
“Open”: c[0]”high”: c[1]”low”: c[2]”close”: c[3],
“Size”: c[4]”time stamp”: c[5]”Merchants”: c[6] or 0,
“Date&Time”: datetime.datetime.fromtimestamp(c[5] /1000),
}

C[5] It is a UNIX millisecond. C[6] It could be nothing.

Step 6: DePIN Combo and Macro

def fetch_depin_listings():
url = f”{CMC_BASE_URL}/v3/cryptocurrency/listings/latest”
Parameters = {“sort”: “volume_24h”, “sort_dir”: “desc”, “limit”: 500, “volume_24h_min”: 500_000}
r = request.get(url, headers=HEADERS, params=params)
r.raise_for_status()
returns r.json()[“data”]

Definition of filter_depin_assets(assets):
Results = []
For assets in assets:
tags = array(asset.get(“tags”) or [])
If tags & DEPIN_TAGS or assets.get(“symbol”) in DEPIN_ASSETS:
results.append(origins)
Return results

def fetch_macro_regime():
fg_url = f”{CMC_BASE_URL}/v3/fear-and-greed/latest”
as_url = f”{CMC_BASE_URL}/v1/altcoin-season-index/latest”
fg = request.get(fg_url, headers=HEADERS).json()[“data”]
tries:
as_idx = request.get(as_url, headers=HEADERS).json()[“data”]
altcoin_index = as_idx.get(“altcoin_index”) or 0 except exception:
altcoin_index = 0
Return {“fear_greed_value”: fg.get(“value”), “altcoin_index”: altcoin_index}

Definition of is_regime_favorable(regime):
return (regime.get(“fear_greed_value”) or 0) > 50 and (regime.get(“altcoin_index”) or 0) >= 50

Step 7: Mass Flow

def run_peaq_monitor(asset_ids, peaq_contract_address=None):
system = fetch_macro_regime()
Raw_quotes = fetch_quotes(list(asset_ids.values()))

quotes = {str(a[“id”]): parse_quote(a) for raw_quotes}
peaq_id = asset_ids.get(“PEAQ”)

peaq_quote = quote.get(str(peaq_id), {})
Peaq_score = compute_peaq_score(peaq_quote)
If not then it is_regime_favorable(regime):
Peak score -= 15

Pool_liq = None
If the peak contract address is:
tries:
swimming pools = fetch_peaq_pools(peaq_contract_address)
best = get_best_pool(pools)
pool_liq = (best or {}).get(“liqUsd”)
Except for the exception:
passes

tries:
lists = fetch_depin_listings()
depin_assets = filter_depin_assets(listings)
Except for the exception:
depin_assets = []

depin_trending = [
{“symbol”: a.get(“symbol”), “pct_24h”: (a.get(“quote”) or [{}])[0].get(“percent_change_24h”)}
To get depin_assets[:10]
]
returns {
“peak_signal”: {
“score”: peaq_score, “price”: peaq_quote.get(“price”),
“pct_24h”: peaq_quote.get(“pct_change_24h”), “volume_24h”: peaq_quote.get(“volume_24h”),
“market_cap”: peaq_quote.get(“market_cap”), “dex_pool_liq”:pool_liq, “regime_favorable”: is_regime_favorable(regime),
},
“depin_trending”: depin_trending,
“System”: the system,
}

Common mistakes

Citation analysis as a dictation in version 3

The quote is a list. Use the following((q for q in assets.get(“quote”, []) if q.get(“symbol”) == “USD”), {}).

Do not pour the liquid dollar to float

liqUsd is a string. Always send: float(p.get(“liqUsd”) or 0).

Pass network_slug only to DEX endpoints

dex_slug is required next to network_slug. Deleting it returns error 400.

Dealing with the CMC as a monitor of machine activity

CMC tracks market prices, not the number of devices or device activity. Use PEAQ’s official API to get real network metrics.

Final thoughts

Key separation:

• CoinMarketCap determines the market conditions and narrative momentum of DePIN’s machine economy
• PEAQ’s official API validates real machine activity and operator rewards

The platforms, protocols, and projects referenced in these articles are used for illustrative purposes only to illustrate how the CoinMarketCap API is integrated in practice. CoinMarketCap makes no representation that these projects endorse or are affiliated with CoinMarketCap. If you are a reference project and have concerns, please contact us at [email protected] and we will respond promptly.