atlas_real_trading.py

Real AsterDEX perpetuals trading coordinator + Flash Scalper skeleton

WARNING: USE AT YOUR OWN RISK — MONEY CAN BE LOST VERY QUICKLY

Paper trade / testnet FIRST. No stop-losses = very high risk.

import os import time import logging from dataclasses import dataclass from typing import List, Optional, Dict from enum import Enum from datetime import datetime

────────────────────────────────────────────────

Dependencies — install these

pip install aster-connector-python python-dotenv tenacity

(or use official aster_dex SDK if you have access)

────────────────────────────────────────────────

try: from aster.rest_api import Client as AsterClient except ImportError: raise ImportError( "Please install aster-connector-python: pip install aster-connector-python\n" "GitHub: https://github.com/asterdex/aster-connector-python" )

from tenacity import retry, stop_after_attempt, wait_exponential

────────────────────────────────────────────────

Logging

────────────────────────────────────────────────

logging.basicConfig( level=logging.INFO, format="%(asctime)s | %(levelname)7s | %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) logger = logging.getLogger("ATLAS-REAL")

────────────────────────────────────────────────

Domain models

────────────────────────────────────────────────

class Direction(Enum): LONG = "BUY" SHORT = "SELL"

@dataclass class Signal: symbol: str direction: Direction confidence: float # 0.0–1.0 expected_profit_usd: float

@dataclass class ProfitTargetConfig: quick_scalp: float = 2.00 standard: float = 2.50 high: float = 3.50 ultra: float = 4.00

@dataclass class AtlasConfig: min_trade_size_usd: float = 30.0 max_trade_size_usd: float = 150.0 max_open_positions: int = 8 leverage: int = 10 cycle_interval_sec: float = 15.0 min_consensus_confidence: float = 0.80 min_confirming_agents: int = 2 watched_symbols: List[str] = None dry_run: bool = True # ← VERY IMPORTANT — set False only when ready

def __post_init__(self):
    if self.watched_symbols is None:
        self.watched_symbols = ["BTCUSDT", "ETHUSDT", "SOLUSDT", "BNBUSDT"]

────────────────────────────────────────────────

Real AsterDEX Exchange Wrapper

────────────────────────────────────────────────

class AsterDEXExchange: def init(self, api_key: str, api_secret: str, testnet: bool = False): base_url = "https://fapi-testnet.asterdex.com" if testnet else "https://fapi.asterdex.com" self.client = AsterClient( key=api_key, secret=api_secret, base_url=base_url, timeout=5, show_limit_usage=True, ) self.testnet = testnet logger.info(f"Connected to AsterDEX {'TESTNET' if testnet else 'MAINNET'}")

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def get_mark_price(self, symbol: str) -> Optional[float]:
    try:
        ticker = self.client.ticker_price(symbol=symbol.upper())
        return float(ticker["price"])
    except Exception as e:
        logger.error(f"Failed to get mark price {symbol}: {e}")
        return None

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=30))
def get_positions(self) -> List[dict]:
    """Returns list of current open positions (simplified)"""
    try:
        return self.client.position()
    except Exception as e:
        logger.error(f"Positions fetch failed: {e}")
        return []

@retry(stop=stop_after_attempt(2), wait=wait_exponential(multiplier=1, min=4, max=15))
def get_balance(self) -> float:
    try:
        bal = self.client.balance()
        for asset in bal:
            if asset["asset"] == "USDT":
                return float(asset.get("availableBalance", 0.0))
        return 0.0
    except Exception as e:
        logger.error(f"Balance fetch failed: {e}")
        return 0.0

def open_market_position(self, symbol: str, side: str, quantity: float, leverage: int, dry_run: bool = True) -> bool:
    if dry_run:
        logger.warning(f"[DRY-RUN] Would open {side} {quantity:.4f} {symbol} @ market (leverage {leverage}x)")
        return True

    try:
        params = {
            "symbol": symbol.upper(),
            "side": side.upper(),
            "type": "MARKET",
            "quantity": f"{quantity:.3f}",  # adjust precision per symbol
            "leverage": str(leverage),
            "positionSide": "BOTH",        # or LONG/SHORT if hedge mode
        }
        resp = self.client.new_order(**params)
        logger.info(f"OPENED {side} {symbol} qty={quantity:.3f} → {resp}")
        return True
    except Exception as e:
        logger.error(f"Open failed {symbol} {side}: {e}")
        return False

def close_position(self, position: dict, dry_run: bool = True) -> bool:
    symbol = position["symbol"]
    side = "SELL" if position["positionAmt"].startswith("+") else "BUY"  # opposite
    qty = abs(float(position["positionAmt"]))

    if dry_run:
        logger.warning(f"[DRY-RUN] Would CLOSE {symbol} {qty:.4f} ({side})")
        return True

    try:
        params = {
            "symbol": symbol,
            "side": side,
            "type": "MARKET",
            "quantity": f"{qty:.3f}",
            "reduceOnly": "true",
        }
        resp = self.client.new_order(**params)
        logger.info(f"CLOSED {symbol} → {resp}")
        return True
    except Exception as e:
        logger.error(f"Close failed {symbol}: {e}")
        return False

────────────────────────────────────────────────

Flash Scalper — IMPLEMENT YOUR REAL LOGIC HERE

────────────────────────────────────────────────

class FlashScalper: def init(self, name: str = "FlashScalper"): self.name = name self.weight = 1.00

def generate_signal(self, symbol: str, exchange: AsterDEXExchange) -> Optional[Signal]:
    """
    IMPLEMENT REAL SIGNAL LOGIC HERE
    Examples:
      - Fetch 1s/5s klines via websocket or REST → momentum / RSI / breakout
      - Use external predictor (ML model, oracle agent, sentiment score)
      - Check order-book imbalance
    """
    # Placeholder — REPLACE completely
    mark_price = exchange.get_mark_price(symbol)
    if not mark_price:
        return None

    # Example dummy condition — YOU MUST CHANGE THIS
    if random.random() > 0.92:  # just ~8% chance — replace with real condition
        direction = Direction.LONG if random.random() > 0.5 else Direction.SHORT
        confidence = round(random.uniform(0.78, 0.96), 3)
        profit_target = round(random.uniform(2.0, 4.2), 2)
        return Signal(symbol, direction, confidence, profit_target)

    return None

────────────────────────────────────────────────

ATLAS CEO — real trading coordinator

────────────────────────────────────────────────

class AtlasCEO: def init( self, config: AtlasConfig, profit_config: ProfitTargetConfig, exchange: AsterDEXExchange, ): self.config = config self.profit_config = profit_config self.exchange = exchange

    # Add real agents when you implement them
    self.workers = [FlashScalper("Flash_v1")]

def collect_signals(self) -> Dict[str, List[Signal]]:
    signals: Dict[str, List[Signal]] = {sym: [] for sym in self.config.watched_symbols}

    for worker in self.workers:
        for symbol in self.config.watched_symbols:
            sig = worker.generate_signal(symbol, self.exchange)
            if sig:
                signals[symbol].append(sig)

    return signals

def aggregate_signals(self, signals_per_symbol: Dict[str, List[Signal]]) -> List[Signal]:
    decisions = []

    for symbol, sigs in signals_per_symbol.items():
        if len(sigs) < self.config.min_confirming_agents:
            continue

        long_score = sum(s.confidence * w for s in sigs if s.direction == Direction.LONG for w in [1.0])
        short_score = sum(s.confidence * w for s in sigs if s.direction == Direction.SHORT for w in [1.0])
        total_w = len(sigs)  # simplified; use real weights later

        if total_w == 0:
            continue

        long_score /= total_w
        short_score /= total_w

        if long_score > short_score and long_score >= self.config.min_consensus_confidence:
            best = max((s for s in sigs if s.direction == Direction.LONG), key=lambda x: x.confidence)
            decisions.append(best)
        elif short_score > long_score and short_score >= self.config.min_consensus_confidence:
            best = max((s for s in sigs if s.direction == Direction.SHORT), key=lambda x: x.confidence)
            decisions.append(best)

    return decisions

def decide_quantity(self, symbol: str, usd_size: float) -> float:
    price = self.exchange.get_mark_price(symbol)
    if not price or price <= 0:
        return 0.0
    # Very naive — in reality use contract size / step size from /exchangeInfo
    qty = usd_size / price
    return round(qty, 3)  # adjust precision per symbol

def check_and_close_winners(self):
    positions = self.exchange.get_positions()
    for pos in positions:
        if float(pos.get("positionAmt", 0)) == 0:
            continue

        unrealized = float(pos.get("unRealizedProfit", 0))
        if unrealized < self.profit_config.quick_scalp:
            continue  # not profitable enough

        logger.info(f"Profitable position detected {pos['symbol']} PnL ${unrealized:.2f}")
        self.exchange.close_position(pos, dry_run=self.config.dry_run)

def run_cycle(self):
    logger.info(f"Cycle {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S UTC')} — balance ~${self.exchange.get_balance():,.1f}")

    self.check_and_close_winners()

    current_positions = len([p for p in self.exchange.get_positions() if float(p.get("positionAmt", 0)) != 0])
    if current_positions >= self.config.max_open_positions:
        logger.info("Max positions reached — skipping entries")
        return

    signals_by_symbol = self.collect_signals()
    final_signals = self.aggregate_signals(signals_by_symbol)

    for sig in final_signals:
        if current_positions >= self.config.max_open_positions:
            break

        size_usd = min(
            max(self.config.min_trade_size_usd, random.uniform(0.7, 1.0) * self.config.max_trade_size_usd),
            self.exchange.get_balance() * 0.15  # max ~15% per trade
        )

        qty = self.decide_quantity(sig.symbol, size_usd)
        if qty <= 0:
            continue

        success = self.exchange.open_market_position(
            symbol=sig.symbol,
            side=sig.direction.value,
            quantity=qty,
            leverage=self.config.leverage,
            dry_run=self.config.dry_run
        )
        if success:
            current_positions += 1

def run(self, cycles: int = 999_999):
    logger.warning("====================================================================")
    logger.warning("          REAL TRADING MODE — MONEY AT RISK — NO STOP LOSSES         ")
    logger.warning(f"                    Dry-run = {self.config.dry_run}                        ")
    logger.warning("====================================================================")

    for i in range(cycles):
        try:
            self.run_cycle()
        except KeyboardInterrupt:
            logger.info("Keyboard interrupt — shutting down")
            break
        except Exception as e:
            logger.exception(f"Cycle crashed: {e}")

        time.sleep(self.config.cycle_interval_sec)

────────────────────────────────────────────────

Main

────────────────────────────────────────────────

if name == "main": from dotenv import load_dotenv load_dotenv()

config = AtlasConfig(
    min_trade_size_usd=35.0,
    max_trade_size_usd=140.0,
    max_open_positions=5,
    leverage=10,
    cycle_interval_sec=12.0,
    min_consensus_confidence=0.78,
    min_confirming_agents=1,   # increase when you have more agents
    dry_run=True,              # ← CHANGE TO FALSE ONLY WHEN YOU ARE READY
)

profit_config = ProfitTargetConfig(
    quick_scalp=1.90,
    standard=2.60,
    high=3.40,
    ultra=4.20,
)

api_key = os.getenv("ASTERDEX_API_KEY")
api_secret = os.getenv("ASTERDEX_API_SECRET")

if not api_key or not api_secret:
    raise ValueError("Missing ASTERDEX_API_KEY or ASTERDEX_API_SECRET in .env")

exchange = AsterDEXExchange(
    api_key=api_key,
    api_secret=api_secret,
    testnet=True   # ← change to False for mainnet
)

ceo = AtlasCEO(config, profit_config, exchange)

ceo.run(cycles=999_999)  # ctrl+c to stop