我使用DeepSeek AI和Dexscreener API构建的一个简单的 AI 加密交易机器人实现了这一目标。在本文中，我将逐步指导您如何构建像我一样的机器人。

DeepSeek 最近发布了R1，这是一种先进的 AI 模型。您可以将其视为 ChatGPT 的免费开源版本，但增加了一些功能。这使它成为构建 AI 驱动应用程序（包括交易机器人）的绝佳工具。Dexscreener 是一种用于跟踪去中心化交易所 (DEX) 的强大工具。它提供有关代币价格、流动性和交易量的实时数据，对加密货币交易者来说必不可少。

步骤 1：入门 首先，在此注册 DeepSeek AI 。

它可以免费使用，并允许您运行自定义提示来构建和部署机器人。 写下你的第一个提示 首先让 DeepSeek AI 充当一位经验丰富的程序员：

“你好。请假装你是一个专业程序员一个哈佛学位和10年的经验。
创造一个机器人到与 Dexscreener 交互。
我们需要到解析、保存和分析每一枚坚固/抽取/成为层的硬币1、上线CEX等到找到模式”。

这将为您的机器人生成初始代码。该机器人将与 Dexscreener API 交互以获取实时数据并进行分析。

第 2 步：添加过滤器和黑名单

加密货币交易存在风险。为了使机器人更安全，我们需要添加过滤器和黑名单。这些功能将：

  1. 过滤掉有风险的代币。
  2. 将与诈骗相关的开发商列入黑名单。

为了使机器人更加强大，我们需要添加过滤器和黑名单以避免有风险的令牌。使用以下提示：

“修改代码 以使用过滤器、硬币黑名单和开发者黑名单将开发人员列入黑名单。
并将所有这些设置放入配置文件中。”

步骤 3：避免虚假音量

具有虚假交易量的代币可能会欺骗您的机器人。要检测虚假交易量，请集成Pocket Universe API。

更新你的机器人

使用以下提示：

“避免使用虚假数量的硬币，通过你的算法检查它是否是假的，或者使用与 Dexscreener 兼容的 Pocket Universe API。”

这会为您的机器人添加一个验证步骤，确保它不会交易不可靠的代币。

步骤 4：防止地毯被拉扯

为了避免诈骗，请与RugCheck.xyz集成。此外，我们可以检测代币供应是否捆绑，这通常是一个危险信号。使用以下提示：

“检查 http://rugcheck.xyz 上的每个代币，并仅与标记为“良好”的合约进行交互。
另外，检查代币的供应是否未捆绑。
如果捆绑 - 黑名单+dev 黑名单。”

第 5 步：自动交易

对于自动交易，请集成BonkBot或Trojan等机器人。这些机器人可以根据您的机器人信号执行交易。 合并代码并添加通知 为了执行交易，我们可以集成现有的交易机器人，如BonkBot或Trojan。我们还将添加 Telegram 通知支持。使用以下提示：

“要交易选定的代币，请通过 Telegram 使用 BonkBot，同时添加 TG 支持以获取买入/卖出通知，并将所有 代码合并为一体。
另外，请写出如何启动此代码。”

第 6 步：启动你的机器人

启动机器人的方法如下： 克隆机器人的存储库（如果您还没有，请向 DeepSeek 索取代码）。 安装依赖项： 

pip 安装 -r 要求.txt txt

config.json使用您的过滤器和 API 密钥配置文件。

运行机器人：

python 交易机器人.py py

步骤 7：可选 — 添加 UI

 最后，您可以要求 DeepSeek 为您的机器人创建 UI。尽管由于流量太大，我无法显示它，但 DeepSeek 成功地为我的机器人创建了一个 UI，使其准备好进行交易和分析来自 Dexscreener 的数据。 询问DeepSeek：

为我的机器人创建一个简单的 Web UI，其中包含过滤器、日志和交易仪表板。

完整代码也可以在这里找到

import requests
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import psycopg2
from sqlalchemy import create_engine
from typing import Dict, List, Optional

# Enhanced Configuration
CONFIG = {
    "DB": {
        "dbname": "dexscreener",
        "user": "admin",
        "password": "your_password",
        "host": "localhost",
        "port": "5432"
    },
    "FILTERS": {
        "min_liquidity": 5000,  # USD
        "min_age_days": 3,
        "coin_blacklist": [
            "0x123...def",  # Known scam token address
            "SUSPECTCOIN"   # Blacklisted symbol
        ],
        "dev_blacklist": [
            "0x456...abc",  # Known rug developer address
            "0x789...fed"   # Another scam developer
        ],
        "chain_whitelist": ["ethereum", "binance-smart-chain"]
    }
}

class EnhancedDexScreenerBot:
    def __init__(self):
        self.engine = create_engine(
            f'postgresql+psycopg2://{CONFIG["DB"]["user"]}:{CONFIG["DB"]["password"]}'
            f'@{CONFIG["DB"]["host"]}/{CONFIG["DB"]["dbname"]}'
        )
        self._init_db()
        self.model = IsolationForest(n_estimators=100, contamination=0.01)
        self.historical_data = self._load_historical_data()

    def _init_db(self):
        """Initialize database with additional security tables"""
        with self.engine.connect() as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS blacklist (
                    address VARCHAR(42) PRIMARY KEY,
                    type VARCHAR(20) CHECK (type IN ('coin', 'dev')),
                    reason TEXT,
                    listed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                );

                CREATE INDEX IF NOT EXISTS idx_blacklist_type ON blacklist(type);
            """)
            # Migrate config blacklists to database
            self._seed_initial_blacklists()

    def _seed_initial_blacklists(self):
        """Initialize blacklists from config"""
        with self.engine.connect() as conn:
            # Seed coin blacklist
            for address in CONFIG["FILTERS"]["coin_blacklist"]:
                conn.execute(
                    """INSERT INTO blacklist (address, type)
                       VALUES (%s, 'coin')
                       ON CONFLICT (address) DO NOTHING""",
                    (address,)
                )
            
            # Seed dev blacklist
            for address in CONFIG["FILTERS"]["dev_blacklist"]:
                conn.execute(
                    """INSERT INTO blacklist (address, type)
                       VALUES (%s, 'dev')
                       ON CONFLICT (address) DO NOTHING""",
                    (address,)
                )

    def apply_filters(self, df: pd.DataFrame) -> pd.DataFrame:
        """Apply all security and quality filters"""
        # Chain whitelist filter
        df = df[df['chain'].isin(CONFIG["FILTERS"]["chain_whitelist"])]
        
        # Liquidity filter
        df = df[df['liquidity'] >= CONFIG["FILTERS"]["min_liquidity"]]
        
        # Age filter
        min_age = datetime.utcnow() - timedelta(days=CONFIG["FILTERS"]["min_age_days"])
        df = df[pd.to_datetime(df['created_at']) < min_age]
        
        # Database blacklist check
        blacklisted_coins = pd.read_sql(
            "SELECT address FROM blacklist WHERE type = 'coin'",
            self.engine
        )['address'].tolist()
        
        blacklisted_devs = pd.read_sql(
            "SELECT address FROM blacklist WHERE type = 'dev'",
            self.engine
        )['address'].tolist()
        
        # Address and symbol checks
        df = df[
            ~df['pair_address'].isin(blacklisted_coins) &
            ~df['base_token_address'].isin(blacklisted_coins) &
            ~df['creator_address'].isin(blacklisted_devs) &
            ~df['base_token_name'].isin(CONFIG["FILTERS"]["coin_blacklist"])
        ]
        
        return df

    def process_data(self, raw_data: List[Dict]) -> pd.DataFrame:
        """Enhanced data processing with security fields"""
        df = pd.DataFrame(raw_data)[[
            'pairAddress', 'baseToken', 'quoteToken', 'priceUsd',
            'liquidity', 'volume', 'chainId', 'dexId', 'createdAt'
        ]]
        
        processed = pd.DataFrame({
            'pair_address': df['pairAddress'],
            'base_token_name': df['baseToken'].apply(lambda x: x['name']),
            'base_token_address': df['baseToken'].apply(lambda x: x['address']),
            'quote_token_address': df['quoteToken'].apply(lambda x: x['address']),
            'price': pd.to_numeric(df['priceUsd']),
            'liquidity': pd.to_numeric(df['liquidity']),
            'volume_24h': pd.to_numeric(df['volume']['h24']),
            'chain': df['chainId'],
            'exchange': df['dexId'],
            'created_at': pd.to_datetime(df['createdAt'], unit='ms'),
            'timestamp': datetime.utcnow()
        })
        
        # Apply security filters
        processed = self.apply_filters(processed)
        
        return processed

    def detect_anomalies(self, new_data: pd.DataFrame) -> pd.DataFrame:
        """Anomaly detection with blacklist awareness"""
        if not new_data.empty:
            features = new_data[['price', 'liquidity', 'volume_24h']]
            features = np.log1p(features)
            
            self.model.fit(self.historical_data)
            anomalies = self.model.predict(features)
            new_data['anomaly_score'] = self.model.decision_function(features)
            return new_data[anomalies == -1]
        return pd.DataFrame()

    def analyze_market_events(self, anomalous_data: pd.DataFrame):
        """Enhanced analysis with blacklist monitoring"""
        for _, row in anomalous_data.iterrows():
            # Check for blacklist pattern matches
            if self._detect_blacklist_pattern(row):
                self._log_event(row, 'BLACKLIST_PATTERN')
            
            # Existing detection logic
            ...

    def _detect_blacklist_pattern(self, row: pd.Series) -> bool:
        """Detect patterns matching known blacklist characteristics"""
        # Check for new addresses similar to blacklisted ones
        similar_coins = pd.read_sql(f"""
            SELECT COUNT(*) FROM blacklist
            WHERE type = 'coin'
            AND similarity(address, '{row['base_token_address']}') > 0.8
            """, self.engine).scalar()
        
        similar_devs = pd.read_sql(f"""
            SELECT COUNT(*) FROM blacklist
            WHERE type = 'dev'
            AND similarity(address, '{row['creator_address']}') > 0.8
            """, self.engine).scalar()
        
        return similar_coins > 0 or similar_devs > 0

    def add_to_blacklist(self, address: str, list_type: str, reason: str):
        """Programmatically add entries to blacklist"""
        with self.engine.connect() as conn:
            conn.execute(
                """INSERT INTO blacklist (address, type, reason)
                   VALUES (%s, %s, %s)
                   ON CONFLICT (address) DO UPDATE SET reason = EXCLUDED.reason""",
                (address, list_type, reason)
            )

    def run(self):
        """Enhanced main loop with filtering"""
        while True:
            try:
                raw_data = self.fetch_pair_data()
                processed_data = self.process_data(raw_data)
                
                if not processed_data.empty:
                    anomalies = self.detect_anomalies(processed_data)
                    self.analyze_market_events(anomalies)
                    
                    processed_data.to_sql(
                        'pairs', self.engine, 
                        if_exists='append', index=False
                    )
                    
                    self.historical_data = pd.concat(
                        [self.historical_data, processed_data]
                    ).tail(100000)
                
                # Update blacklists periodically
                self._refresh_blacklists()
                time.sleep(60)  # Add sleep between iterations

            except Exception as e:
                print(f"Runtime error: {e}")

    def _refresh_blacklists(self):
        """Refresh blacklists from external sources"""
        # Example: Sync with community-maintained blacklists
        try:
            response = requests.get("https://api.gopluslabs.io/api/v1/token_security/1")
            data = response.json()
            for token in data['tokens']:
                if token['is_honeypot']:
                    self.add_to_blacklist(
                        token['contract_address'], 
                        'coin', 
                        'Automated honeypot detection'
                    )
        except Exception as e:
            print(f"Blacklist refresh failed: {e}")

# Example usage with blacklist management
if __name__ == "__main__":
    bot = EnhancedDexScreenerBot()
    
    # Manually add suspicious entry
    bot.add_to_blacklist(
        "0xNEW...SCAM", 
        "dev", 
        "Suspicious deployment pattern"
    )
    
    bot.run()

1

通过遵循这些步骤，您可以创建一个实时的加密货币交易机器人，该机器人可以：
足够聪明，可以避免诈骗。
能够有效地检测趋势和模式。

如果我能在一夜之间将100美元变成35,000美元，你也可以！试试看，祝你交易愉快！🚀

免责声明：加密货币交易涉及高风险。此机器人是协助您分析的工具，不是财务建议。请始终负责任地进行交易。 

1

# 人工智能 # python
Logo
DeepSeek技术社区

欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。

更多推荐

cover

DeepSeek本地部署全攻略：保姆级教程带你轻松上手

avatar DeepSeek技术社区
cover

复旦NLP团队2025新书《大规模语言模型：从理论到实践（第2版）》重磅来袭！

avatar DeepSeek技术社区
cover

不可错过！DeepSeek + 本地知识库，打造超强AI助手

avatar DeepSeek技术社区