Cryptocurrency markets operate continuously across multiple trading venues, creating a large amount of market data that can be difficult to evaluate manually. A systematic screening process can help traders identify assets with strong momentum, evaluate technical trend confirmation, consider upcoming crypto events, and calculate risk-based trading levels before considering a position.
This project develops a Crypto Trading Intelligence System that combines MarketFlow market intelligence with cryptocurrency OHLCV data. The system retrieves strong-trend cryptocurrency data, identifies upcoming crypto events, normalizes the datasets, evaluates hourly technical conditions, calculates ATR-based risk levels, and produces a final trading intelligence report.
The system is configured for a 1-hour timeframe, up to 100 cryptocurrencies, 1,000 USDT simulated capital, 1% risk per trade, an ATR multiplier of 1.5, and a 2.0 reward-to-risk ratio. The event analysis uses a three-day lookahead period. crypto_trading_intelligence_mul…
Cell 1 — Crypto Trading Intelligence Configuration
The first cell establishes the Python environment and MarketFlow API connection. It imports Requests for HTTP communication, NumPy and Pandas for data processing, Matplotlib for visualization, and IPython Display for DataFrame presentation.
The trading configuration defines the analysis timeframe, candle limit, maximum number of cryptocurrencies, ranking size, capital, risk percentage, ATR multiplier, reward-to-risk ratio, event lookahead period, and API timeout.
The code also performs an API connection test against the MarketFlow strong-trend endpoint and provides basic HTTP status handling for authentication, access, and rate-limit problems. crypto_trading_intelligence_mul…
Code
# ============================================================
# CELL 1/5 — CRYPTO TRADING INTELLIGENCE CONFIGURATION
# ============================================================
import os
import re
import time
import getpass
from datetime import datetime, timedelta, timezone
import requests
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from IPython.display import display
# ============================================================
# 1. RAPIDAPI CONFIGURATION
# ============================================================
RAPIDAPI_KEY = os.getenv("RAPIDAPI_KEY", "").strip()
if not RAPIDAPI_KEY:
RAPIDAPI_KEY = getpass.getpass(
"Enter your RapidAPI Key: "
).strip()
if not RAPIDAPI_KEY:
raise ValueError(
"RapidAPI Key wajib diisi. "
"Dapatkan key dari dashboard RapidAPI."
)
API_HOST = (
"marketflow-all-in-one-market-finance-api"
".p.rapidapi.com"
)
BASE_URL = f"https://{API_HOST}"
HEADERS = {
"x-rapidapi-key": RAPIDAPI_KEY,
"x-rapidapi-host": API_HOST
}
SESSION = requests.Session()
# ============================================================
# 2. TRADING CONFIGURATION
# ============================================================
TIMEFRAME = "1h"
CANDLE_LIMIT = 220
MAX_COINS = 100
TOP_N = 12
CAPITAL_USDT = 1000.0
RISK_PER_TRADE = 0.01
ATR_MULTIPLIER = 1.5
REWARD_RISK_RATIO = 2.0
EVENT_LOOKAHEAD_DAYS = 3
REQUEST_TIMEOUT = 25
NOW = datetime.now(timezone.utc)
# ============================================================
# 3. API CONNECTION TEST
# ============================================================
print("=" * 65)
print("CRYPTO TRADING INTELLIGENCE — API CONNECTION")
print("=" * 65)
try:
response = SESSION.get(
BASE_URL + "/crypto/strong-trend",
headers=HEADERS,
timeout=REQUEST_TIMEOUT
)
print("HTTP Status:", response.status_code)
if response.status_code == 200:
print("MarketFlow API: CONNECTED")
elif response.status_code == 401:
print("ERROR: Invalid RapidAPI Key")
elif response.status_code == 403:
print("ERROR: API subscription/access denied")
elif response.status_code == 429:
print("ERROR: API rate limit exceeded")
else:
print("API Response:", response.text[:300])
except requests.RequestException as error:
print("Connection Error:", str(error))
# ============================================================
# 4. CONFIGURATION SUMMARY
# ============================================================
print("\n" + "=" * 65)
print("TRADING SYSTEM CONFIGURATION")
print("=" * 65)
print("Analysis Time :", NOW)
print("Timeframe :", TIMEFRAME)
print("Capital :", CAPITAL_USDT, "USDT")
print("Risk/Trade :", RISK_PER_TRADE * 100, "%")
print("Risk/Reward :", REWARD_RISK_RATIO)
print("ATR Multiplier:", ATR_MULTIPLIER)
print("Max Coins :", MAX_COINS)
print("Event Window :", EVENT_LOOKAHEAD_DAYS, "days")
print("=" * 65)
The configuration establishes the analytical assumptions used by all subsequent cells. In particular, the 1% risk allocation means the nominal risk budget is:
\[ Risk\ Budget = Capital \times Risk\ Per\ Trade \]
With the configured parameters:
\[ Risk\ Budget = 1000 \times 0.01 = 10\ USDT \]
This value is later used in the position-sizing calculation.
Cell 2 — MarketFlow API Data Collection
The second cell creates a reusable API function and retrieves two major datasets.
The first dataset comes from /crypto/strong-trend and provides the market trend information used as the initial universe for the screener.
The second dataset comes from /crypto-events. The query is restricted to reliable sources and covers the current date through the next three days. The source code explicitly warns that event pagination begins at skip=0, meaning additional event pages may exist. crypto_trading_intelligence_mul…
Code
# CELL 2/5 — MARKETFLOW DATA COLLECTION
def get_json(endpoint, params=None):
response = SESSION.get(
BASE_URL + endpoint,
headers=HEADERS,
params=params,
timeout=REQUEST_TIMEOUT
)
if response.status_code != 200:
raise RuntimeError(
f"{endpoint}: HTTP {response.status_code} "
f"{response.text[:300]}"
)
data = response.json()
if isinstance(data, dict):
if data.get("success") is False or data.get("error"):
raise RuntimeError(str(data)[:400])
return data
START_DATE = NOW.date().isoformat()
END_DATE = (
NOW.date() + timedelta(days=EVENT_LOOKAHEAD_DAYS)
).isoformat()
# ==================================================
# STRONG TREND
# ==================================================
strong_raw = get_json(
"/crypto/strong-trend"
)
# ==================================================
# CRYPTO EVENTS
# ==================================================
events_raw = get_json(
"/crypto-events",
params={
"skip": 0,
"source_reliable": "true",
"important": "false",
"start_date": START_DATE,
"end_date": END_DATE
}
)
print("=" * 65)
print("MARKETFLOW API COLLECTION REPORT")
print("=" * 65)
print("Strong Trend :", type(strong_raw).__name__)
print("Crypto Events:", type(events_raw).__name__)
print("Event Window :", START_DATE, "to", END_DATE)
print("\nSTRONG TREND RESPONSE SAMPLE")
print(str(strong_raw)[:800])
print("\nCRYPTO EVENTS RESPONSE SAMPLE")
print(str(events_raw)[:800])
print("\nWARNING: Event pagination starts at skip=0.")
print("Additional pages may exist.")
The purpose of this cell is to separate market trend intelligence from event intelligence. This is useful because an asset can have a technically strong trend while simultaneously having an upcoming event that increases short-term uncertainty.
Cell 3 — Data Cleaning & Event Intelligence
The third cell converts the raw API responses into standardized datasets.
The extract_rows() function handles different possible JSON structures. The clean_symbol() function normalizes cryptocurrency identifiers and removes common USDT and USD suffixes.
Trend classification prioritizes explicit API trend labels. If a valid label is unavailable, the system falls back to 24-hour price change. A change above 5% is classified as bullish, while a change below -5% is classified as bearish. Intermediate values are classified as neutral. crypto_trading_intelligence_mul…
The event-processing section maps each event to its cryptocurrency and converts the event date to UTC. Events marked as unreliable are excluded. crypto_trading_intelligence_mul…
Code
# CELL 3/5 — DATA CLEANING & EVENT INTELLIGENCE
def extract_rows(obj, preferred=None):
preferred = preferred or []
if isinstance(obj, list):
return [x for x in obj if isinstance(x, dict)]
if not isinstance(obj, dict):
return []
keys = preferred + [
"data", "results", "items",
"events", "trends", "coins",
"cryptocurrencies"
]
for key in keys:
if key in obj:
result = extract_rows(obj[key], preferred)
if result:
return result
for value in obj.values():
if isinstance(value, list):
result = extract_rows(value, [])
if result:
return result
return []
def get_value(record, *keys):
if not isinstance(record, dict):
return None
lookup = {
str(k).lower(): v
for k, v in record.items()
}
for key in keys:
value = lookup.get(key.lower())
if value is not None and value != "":
return value
return None
def clean_symbol(value):
if isinstance(value, dict):
value = get_value(
value, "symbol", "ticker", "code"
)
symbol = re.sub(
r"[^A-Z0-9]", "",
str(value or "").upper()
)
for suffix in ["USDT", "USD"]:
if symbol.endswith(suffix) and len(symbol) > len(suffix):
symbol = symbol[:-len(suffix)]
break
return symbol if 2 <= len(symbol) <= 15 else ""
def to_number(value):
try:
if isinstance(value, str):
value = value.replace("%", "").replace(",", "")
return float(value)
except (ValueError, TypeError):
return np.nan
def classify_trend(record):
label = str(get_value(
record,
"classification", "trend",
"direction", "signal",
"trend_type", "category", "status"
) or "").lower()
if any(x in label for x in [
"bearish", "downtrend", "strong_bear"
]):
return "BEARISH"
if any(x in label for x in [
"bullish", "uptrend", "strong_bull"
]):
return "BULLISH"
change = to_number(get_value(
record,
"price_change_percentage_24h",
"price_change_24h_percent",
"change_24h_percent",
"change_24h",
"price_change_24h"
))
if np.isfinite(change):
if change > 5:
return "BULLISH"
if change < -5:
return "BEARISH"
return "NEUTRAL"
return "UNKNOWN"
# ==================================================
# NORMALIZE STRONG TREND
# ==================================================
trend_records = extract_rows(
strong_raw,
["strong_trends", "strong_trend",
"strong_bullish", "strong_bearish"]
)
trends = []
for record in trend_records:
coin = clean_symbol(get_value(
record,
"symbol", "ticker", "coin",
"asset", "id", "name"
))
if not coin:
continue
change = to_number(get_value(
record,
"price_change_percentage_24h",
"price_change_24h_percent",
"change_24h_percent",
"change_24h",
"price_change_24h"
))
trends.append({
"coin": coin,
"trend": classify_trend(record),
"change_24h_pct": change
})
trend_df = pd.DataFrame(
trends,
columns=["coin", "trend", "change_24h_pct"]
).drop_duplicates("coin")
# ==================================================
# NORMALIZE CRYPTO EVENTS
# ==================================================
event_records = extract_rows(
events_raw,
["events", "crypto_events", "calendar"]
)
events = []
for record in event_records:
coin = clean_symbol(get_value(
record,
"symbol", "ticker", "coin",
"asset", "currency"
))
date = pd.to_datetime(
get_value(
record, "date", "event_date",
"start_date", "datetime", "timestamp"
),
utc=True,
errors="coerce"
)
if not coin or pd.isna(date):
continue
reliable = get_value(
record, "source_reliable", "reliable"
)
if str(reliable).lower() == "false":
continue
title = str(get_value(
record,
"title", "name", "event", "description"
) or "")
events.append({
"coin": coin,
"event_utc": date,
"event_title": title[:110]
})
events_df = pd.DataFrame(
events,
columns=["coin", "event_utc", "event_title"]
)
print("=" * 65)
print("CRYPTO MARKET INTELLIGENCE")
print("=" * 65)
print("Coins detected :", len(trend_df))
print("Events mapped :", len(events_df))
print("Raw events :", len(event_records))
display(trend_df.head(15))
display(events_df.head(10))
if trend_df.empty:
print("WARNING: Periksa skema strong_raw.")
This stage is particularly important because raw API responses should not be sent directly into the trading engine. The system first converts the data into a consistent structure:
cointrendchange_24h_pct
for market trends, and:
coinevent_utcevent_title
for event intelligence.
This normalization also makes the later multi-exchange OHLCV process independent of the exact MarketFlow response structure.
Cell 4 — Technical Analysis & Trading Signals
The source file contains two consecutive technical-analysis sections labeled Cell 4. The first section uses Binance directly, while the second section implements the multi-exchange engine with KuCoin as primary and Gate.io as backup. Both are retained below because they are part of the source code. crypto_trading_intelligence_mul…
The first implementation retrieves hourly Binance candles and calculates:
EMA20
EMA50
ATR14
24-hour momentum
A bullish setup requires:
\[ Trend = BULLISH \]
and:
\[ Price > EMA20 > EMA50 \]
A bearish setup requires:
\[ Trend = BEARISH \]
and:
\[ Price < EMA20 < EMA50 \]
An upcoming event overrides the technical setup and leaves the asset in a waiting state.
The stop-loss distance is:
\[ Distance = ATR_{14} \times ATR\ Multiplier \]
and the take-profit distance is:
\[ TP\ Distance = Distance \times Reward/Risk \]
Position quantity is then constrained by both the risk budget and available capital. crypto_trading_intelligence_mul…
The second Cell 4 is the multi-exchange implementation and is the most important part of the uploaded system.
It uses:
Primary exchange: KuCoin
Backup exchange: Gate.io
The purpose of this architecture is to prevent the entire screening process from depending on one OHLCV provider. If KuCoin cannot provide valid candles, the system attempts Gate.io.
The OHLCV validation layer checks:
minimum candle count
closed candles
data freshness
hourly continuity
positive OHLCV values
logical high/low relationships
The technical engine then calculates EMA20, EMA50, ATR14, and 24-hour price change. crypto_trading_intelligence_mul…
Code
# CELL 4/5 — TECHNICAL ANALYSIS & TRADING SIGNALS
def get_candles(coin):
response = SESSION.get(
"https://api.binance.com/api/v3/klines",
params={
"symbol": coin + "USDT",
"interval": TIMEFRAME,
"limit": CANDLE_LIMIT
},
timeout=REQUEST_TIMEOUT
)
response.raise_for_status()
candles = response.json()
if not isinstance(candles, list) or len(candles) < 80:
raise ValueError("Insufficient candle data")
df = pd.DataFrame(candles).iloc[:, :7]
df.columns = [
"open_time", "open", "high",
"low", "close", "volume", "close_time"
]
for column in [
"open", "high", "low", "close", "volume"
]:
df[column] = pd.to_numeric(
df[column], errors="coerce"
)
df["close_time"] = pd.to_datetime(
df["close_time"], unit="ms", utc=True
)
df = df[
df["close_time"] <= pd.Timestamp.now(tz="UTC")
].copy()
if len(df) < 80:
raise ValueError("Insufficient closed candles")
df["ema20"] = df["close"].ewm(
span=20, adjust=False
).mean()
df["ema50"] = df["close"].ewm(
span=50, adjust=False
).mean()
previous_close = df["close"].shift(1)
tr = pd.concat([
df["high"] - df["low"],
(df["high"] - previous_close).abs(),
(df["low"] - previous_close).abs()
], axis=1).max(axis=1)
df["atr14"] = tr.ewm(
alpha=1 / 14, adjust=False
).mean()
return df.iloc[-1]
# ==================================================
# GENERATE TRADING SETUPS
# ==================================================
results = []
for _, row in trend_df.head(MAX_COINS).iterrows():
coin = row["coin"]
trend = row["trend"]
matching = events_df[
events_df["coin"] == coin
]
upcoming = matching[
(matching["event_utc"] >= pd.Timestamp(NOW)) &
(
matching["event_utc"] <=
pd.Timestamp(
NOW + timedelta(days=EVENT_LOOKAHEAD_DAYS)
)
)
]
event_risk = len(upcoming) > 0
item = {
"Coin": coin,
"Trend": trend,
"24h %": row["change_24h_pct"],
"Event <=3d": len(upcoming),
"Event Detail": " | ".join(
upcoming["event_title"].head(2)
),
"Status": "WAIT",
"Reason": "",
"Price": np.nan,
"EMA20": np.nan,
"EMA50": np.nan,
"ATR14": np.nan,
"Entry": np.nan,
"SL": np.nan,
"TP1 (2R)": np.nan,
"Qty": np.nan,
"Notional USDT": np.nan,
"Risk USDT": np.nan,
"Candle UTC": ""
}
try:
candle = get_candles(coin)
price = float(candle["close"])
atr = float(candle["atr14"])
ema20 = float(candle["ema20"])
ema50 = float(candle["ema50"])
item.update({
"Price": price,
"EMA20": ema20,
"EMA50": ema50,
"ATR14": atr,
"Candle UTC": str(candle["close_time"])
})
long_ok = (
trend == "BULLISH" and
price > ema20 > ema50
)
short_ok = (
trend == "BEARISH" and
price < ema20 < ema50
)
if event_risk:
item["Reason"] = "Upcoming event risk"
elif not (long_ok or short_ok):
item["Reason"] = "EMA confirmation failed"
elif not np.isfinite(atr) or atr <= 0:
item["Reason"] = "Invalid ATR"
else:
side = 1 if long_ok else -1
stop_loss = (
price - side * ATR_MULTIPLIER * atr
)
take_profit = (
price +
side * ATR_MULTIPLIER *
atr * REWARD_RISK_RATIO
)
risk_per_unit = abs(price - stop_loss)
risk_budget = (
CAPITAL_USDT * RISK_PER_TRADE
)
quantity = risk_budget / risk_per_unit
quantity = min(
quantity, CAPITAL_USDT / price
)
item.update({
"Status": (
"LONG SETUP" if long_ok
else "SHORT SETUP"
),
"Reason": "Trend and EMA confirmed",
"Entry": price,
"SL": stop_loss,
"TP1 (2R)": take_profit,
"Qty": quantity,
"Notional USDT": quantity * price,
"Risk USDT": (
quantity * risk_per_unit
)
})
except (
requests.RequestException,
ValueError,
TypeError,
KeyError,
RuntimeError
) as error:
item["Reason"] = (
"OHLCV unavailable: " + str(error)[:75]
)
results.append(item)
report = pd.DataFrame(results)
if not report.empty:
report["Priority"] = np.where(
report["Status"].isin([
"LONG SETUP", "SHORT SETUP"
]),
1, 2
)
report["Momentum Rank"] = (
report["24h %"].abs().fillna(-1)
)
report = report.sort_values(
["Priority", "Momentum Rank"],
ascending=[True, False]
).reset_index(drop=True)
print("Technical analysis completed.")
print("No automatic orders will be executed.")
The multi-exchange implementation adds an important operational feature: provider fallback. The function get_candles() tries KuCoin first and Gate.io second. If both providers fail, the system returns the failure as a data-quality issue instead of inventing a trading signal. crypto_trading_intelligence_mul…
The position-sizing logic is:
\[ Position\ Quantity = \min \left( \frac{Risk\ Budget}{ATR\ Distance}, \frac{Capital}{Entry} \right) \]
This means the system considers both the desired monetary risk and the maximum position affordable under the configured capital.
Cell 5 — Final Trading Intelligence Report
The final cell presents the ranked trading candidates, executive summary, confirmed setups, momentum chart, CSV export, and risk disclosure.
The report prioritizes actual LONG SETUP and SHORT SETUP conditions over WAIT conditions. Within each priority group, the system uses absolute 24-hour momentum as a secondary ranking variable. crypto_trading_intelligence_mul…
The final report includes:
Coin
Trend
24-hour change
Event status
Status
Price
Entry
Stop loss
Take profit
Quantity
Risk
Reason
The executive summary reports the total number of coins, LONG setups, SHORT setups, WAIT conditions, risk budget, and reward-to-risk ratio. crypto_trading_intelligence_mul…
Code
# CELL 5/5 — FINAL TRADING INTELLIGENCE REPORT
print("=" * 75)
print("CRYPTO — FINAL TRADING INTELLIGENCE REPORT")
print("=" * 75)
if report.empty:
print("No valid candidates detected.")
print("Check MarketFlow response structure.")
else:
display_columns = [
"Coin",
"Trend",
"24h %",
"Event <=3d",
"Status",
"Price",
"Entry",
"SL",
"TP1 (2R)",
"Qty",
"Risk USDT",
"Reason"
]
print("\nTOP TRADING CANDIDATES")
display(
report[display_columns]
.head(TOP_N)
.style.format(
{
"Price": "{:,.6f}",
"Entry": "{:,.6f}",
"SL": "{:,.6f}",
"TP1 (2R)": "{:,.6f}",
"Qty": "{:,.6f}",
"Risk USDT": "{:,.2f}"
},
na_rep="-"
)
)
# ==================================================
# EXECUTIVE SUMMARY
# ==================================================
counts = report["Status"].value_counts()
print("\n" + "=" * 75)
print("TRADING DESK EXECUTIVE SUMMARY")
print("=" * 75)
print(
"Timestamp UTC:",
datetime.now(timezone.utc).strftime(
"%Y-%m-%d %H:%M"
)
)
print("Total Coins :", len(report))
print(
"LONG Setups :",
counts.get("LONG SETUP", 0)
)
print(
"SHORT Setups:",
counts.get("SHORT SETUP", 0)
)
print(
"WAIT :",
counts.get("WAIT", 0)
)
print(
"Risk Budget :",
CAPITAL_USDT * RISK_PER_TRADE,
"USDT"
)
print(
"Risk/Reward :",
REWARD_RISK_RATIO
)
# ==================================================
# TRADEABLE SETUPS
# ==================================================
actionable = report[
report["Status"].isin([
"LONG SETUP",
"SHORT SETUP"
])
]
print("\nCONFIRMED TECHNICAL SETUPS")
if actionable.empty:
print("NO TRADE — No confirmed setups.")
else:
display(
actionable[display_columns].head(TOP_N)
)
# ==================================================
# MOMENTUM CHART
# ==================================================
chart_data = report.head(TOP_N)
if chart_data["24h %"].notna().any():
ax = chart_data.set_index("Coin")[
"24h %"
].plot(
kind="bar",
figsize=(12, 5),
title="Crypto Strong Trend — 24H Momentum",
ylabel="Price Change (%)"
)
plt.axhline(
0, color="gray", linewidth=1
)
plt.xticks(rotation=45)
plt.tight_layout()
plt.show()
# ==================================================
# EXPORT REPORT
# ==================================================
filename = (
"crypto_trading_intelligence_" +
datetime.now(timezone.utc).strftime(
"%Y%m%d_%H%M"
) +
".csv"
)
report.drop(
columns=["Priority", "Momentum Rank"]
).to_csv(
filename,
index=False
)
print("\nREPORT EXPORTED:", filename)
print("\nRISK DISCLOSURE")
print("Technical setup is not a guaranteed trade.")
print("Check live price, liquidity, spread and fees.")
print("SHORT requires a compatible derivatives venue.")
print("Unmapped events do not imply zero event risk.")
The final output deliberately distinguishes between screening candidates and confirmed technical setups. If no candidate satisfies the required rules, the system returns:
NO TRADE — No confirmed setups.That behavior is important because a screener should not force a trade simply because the market contains assets with positive or negative momentum.
The code also explicitly states that no automatic orders are submitted.
Conclusion
This Crypto Trading Intelligence System provides a structured framework for screening cryptocurrency markets using market trend intelligence, upcoming crypto events, hourly OHLCV data, EMA20, EMA50, ATR14, momentum, and volatility-adjusted position sizing.
The workflow begins with MarketFlow API configuration and market-data collection, then normalizes the strong-trend and event datasets before applying technical analysis. The multi-exchange implementation uses KuCoin as the primary OHLCV provider and Gate.io as the fallback, while validating candle continuity, data freshness, and OHLCV consistency before producing a trading setup. crypto_trading_intelligence_mul…
The trading logic requires alignment between the MarketFlow trend classification and EMA structure. A bullish candidate must satisfy BULLISH with Price > EMA20 > EMA50, while a bearish candidate must satisfy the inverse condition. Upcoming mapped events can prevent a setup from becoming actionable. Stop loss and take profit are then derived from ATR and the predefined reward-to-risk ratio.
The system should therefore be treated as a trading screening and decision-support framework, not as a guaranteed predictive model. The source code itself highlights several limitations: event pagination is incomplete from the initial skip=0 request, unmapped events do not imply zero event risk, and live execution conditions such as liquidity, spread, and fees still need to be checked. crypto_trading_intelligence_mul… crypto_trading_intelligence_mul…
The strongest application of this framework is to use it as the first layer of a quantitative trading workflow. Assets identified by the screener should receive additional confirmation through market structure, liquidity, volume, support and resistance, higher-timeframe trend, and execution conditions before any live position is considered.