Python для крипторынка: автоматизация торговли и анализ данных

import ccxt

# Инициализация объекта биржи
exchange = ccxt.binance({
'apiKey': 'ВАШ_API_КЛЮЧ',
'secret': 'ВАШ_СЕКРЕТНЫЙ_КЛЮЧ',
})

# Получение баланса
balance = exchange.fetch_balance()
print(balance['BTC'])

from binance.websocket.spot.websocket_client import SpotWebsocketClient

def message_handler(message):
print(message)

ws_client = SpotWebsocketClient()
ws_client.start()

# Подписка на канал котировок BTC/USDT
ws_client.ticker(symbol="btcusdt", callback=message_handler)

# Держим соединение активным
import time
time.sleep(10)
ws_client.stop()

import pandas as pd
import talib
import matplotlib.pyplot as plt
import ccxt

# Получение исторических данных
exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv('BTC/USDT', '1h', limit=500)

# Преобразование в DataFrame
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)

# Расчет индикаторов
df['sma_20'] = talib.SMA(df['close'].values, timeperiod=20)
df['sma_50'] = talib.SMA(df['close'].values, timeperiod=50)
df['rsi'] = talib.RSI(df['close'].values, timeperiod=14)

# Определение сигналов
df['signal'] = 0
df.loc[df['sma_20'] > df['sma_50'], 'signal'] = 1
df.loc[df['sma_20'] < df['sma_50'], 'signal'] = -1

# Визуализация
plt.figure(figsize=(12, 8))
plt.subplot(2, 1, 1)
plt.plot(df.index, df['close'], label='Close Price')
plt.plot(df.index, df['sma_20'], label='SMA 20')
plt.plot(df.index, df['sma_50'], label='SMA 50')
plt.legend()

plt.subplot(2, 1, 2)
plt.plot(df.index, df['rsi'], label='RSI')
plt.axhline(y=70, color='r', linestyle='--')
plt.axhline(y=30, color='g', linestyle='--')
plt.legend()

plt.tight_layout()
plt.show()

from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score

# Подготовка признаков
df['return'] = df['close'].pct_change()
df['target'] = (df['return'].shift(-1) > 0).astype(int) # 1, если цена вырастет, 0 — если упадет

# Добавление лагов и технических индикаторов в качестве признаков
for i in range(1, 6):
df[f'return_lag_{i}'] = df['return'].shift(i)
df[f'volume_lag_{i}'] = df['volume'].shift(i)

df['volatility'] = df['close'].rolling(window=20).std()
df.dropna(inplace=True)

# Разделение на признаки и целевую переменную
features = ['return', 'volume', 'rsi', 'volatility'] + [f'return_lag_{i}' for i in range(1, 6)] + [f'volume_lag_{i}' for i in range(1, 6)]
X = df[features]
y = df['target']

# Разделение на обучающую и тестовую выборки
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, shuffle=False)

# Обучение модели
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)

# Оценка точности
predictions = model.predict(X_test)
accuracy = accuracy_score(y_test, predictions)
print(f'Точность модели: {accuracy:.2f}')

# Важность признаков
feature_importance = pd.DataFrame({
'feature': features,
'importance': model.feature_importances_
}).sort_values('importance', ascending=False)
print(feature_importance)

import ccxt
import pandas as pd
import numpy as np
import time
from datetime import datetime
import logging

# Настройка логирования
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s – %(name)s – %(levelname)s – %(message)s',
filename='trading_bot.log'
)
logger = logging.getLogger('TradingBot')

# Конфигурация бота
config = {
'exchange': 'binance',
'symbol': 'BTC/USDT',
'apiKey': 'ВАШ_API_КЛЮЧ',
'secret': 'ВАШ_СЕКРЕТНЫЙ_КЛЮЧ',
'timeframe': '15m',
'fast_period': 10,
'slow_period': 30,
'amount_to_trade': 0.001, # BTC
'max_positions': 1
}

# Инициализация биржи
exchange = getattr(ccxt, config['exchange'])({
'apiKey': config['apiKey'],
'secret': config['secret'],
'enableRateLimit': True
})

def get_historical_data():
"""Получение исторических данных для анализа"""
ohlcv = exchange.fetch_ohlcv(
config['symbol'],
config['timeframe'],
limit=100 # Достаточно для расчета индикаторов
)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)

# Расчет скользящих средних
df['fast_ma'] = df['close'].rolling(window=config['fast_period']).mean()
df['slow_ma'] = df['close'].rolling(window=config['slow_period']).mean()

return df

def generate_signals(df):
"""Генерация торговых сигналов"""
df['signal'] = 0
# Сигнал на покупку: быстрая MA пересекает медленную MA снизу вверх
df.loc[(df['fast_ma'] > df['slow_ma']) & (df['fast_ma'].shift(1) <= df['slow_ma'].shift(1)), 'signal'] = 1
# Сигнал на продажу: быстрая MA пересекает медленную MA сверху вниз
df.loc[(df['fast_ma'] < df['slow_ma']) & (df['fast_ma'].shift(1) >= df['slow_ma'].shift(1)), 'signal'] = -1

return df

def check_positions():
"""Проверка текущих позиций"""
balance = exchange.fetch_balance()
return float(balance['BTC']['free'])

def execute_trade(signal):
"""Исполнение торговой операции"""
current_positions = check_positions()

if signal == 1 and current_positions < config['max_positions'] * config['amount_to_trade']:
# Размещение ордера на покупку
try:
order = exchange.create_market_buy_order(
config['symbol'],
config['amount_to_trade']
)
logger.info(f"BUY ORDER: {order}")
return True
except Exception as e:
logger.error(f"Error creating buy order: {str(e)}")
return False

elif signal == -1 and current_positions > 0:
# Размещение ордера на продажу
try:
order = exchange.create_market_sell_order(
config['symbol'],
min(current_positions, config['amount_to_trade'])
)
logger.info(f"SELL ORDER: {order}")
return True
except Exception as e:
logger.error(f"Error creating sell order: {str(e)}")
return False

return False

def trading_loop():
"""Основной цикл торгового бота"""
logger.info("Starting trading bot...")

while True:
try:
# Получение и анализ данных
df = get_historical_data()
df = generate_signals(df)

# Получение последнего сигнала
latest_signal = df['signal'].iloc[-1]

# Исполнение сигнала, если он есть
if latest_signal != 0:
success = execute_trade(latest_signal)
if success:
logger.info(f"Successfully executed signal: {latest_signal}")
else:
logger.warning(f"Failed to execute signal: {latest_signal}")

# Ожидание до следующей проверки
time.sleep(60) # Проверка каждую минуту

except Exception as e:
logger.error(f"Error in trading loop: {str(e)}")
time.sleep(300) # При ошибке ждем 5 минут перед повторной попыткой

if __name__ == "__main__":
trading_loop()

from web3 import Web3
import json
from solcx import compile_standard, install_solc

# Установка компилятора Solidity
install_solc('0.8.0')

# Смарт-контракт на Solidity
contract_source = '''
// SPDX-License-Identifier: MIT
pragma solidity ^0.8.0;

contract SimpleToken {
string public name = "SimpleToken";
string public symbol = "STK";
uint256 public totalSupply = 1000000;
mapping(address => uint256) public balanceOf;

event Transfer(address indexed from, address indexed to, uint256 value);

constructor() {
balanceOf[msg.sender] = totalSupply;
}

function transfer(address to, uint256 value) public returns (bool success) {
require(balanceOf[msg.sender] >= value, "Insufficient balance");

balanceOf[msg.sender] -= value;
balanceOf[to] += value;

emit Transfer(msg.sender, to, value);
return true;
}
}
'''

# Компиляция смарт-контракта
compiled_sol = compile_standard(
{
"language": "Solidity",
"sources": {"SimpleToken.sol": {"content": contract_source}},
"settings": {
"outputSelection": {
"*": {"*": ["abi", "metadata", "evm.bytecode", "evm.sourceMap"]}
}
},
},
solc_version="0.8.0",
)

# Получение байткода и ABI контракта
contract_bytecode = compiled_sol["contracts"]["SimpleToken.sol"]["SimpleToken"]["evm"]["bytecode"]["object"]
contract_abi = compiled_sol["contracts"]["SimpleToken.sol"]["SimpleToken"]["abi"]

# Подключение к блокчейну (локальный Ganache для тестирования)
w3 = Web3(Web3.HTTPProvider("http://127.0.0.1:8545"))

# Аккаунт для развертывания
deployer_account = w3.eth.accounts[0]
w3.eth.default_account = deployer_account

# Создание контракта
SimpleToken = w3.eth.contract(abi=contract_abi, bytecode=contract_bytecode)

# Развертывание контракта
tx_hash = SimpleToken.constructor().transact()
tx_receipt = w3.eth.wait_for_transaction_receipt(tx_hash)

# Получение адреса развернутого контракта
contract_address = tx_receipt.contractAddress
print(f"Контракт развернут по адресу: {contract_address}")

# Создание экземпляра контракта для взаимодействия
token = w3.eth.contract(address=contract_address, abi=contract_abi)

# Проверка баланса владельца
owner_balance = token.functions.balanceOf(deployer_account).call()
print(f"Баланс владельца: {owner_balance} STK")

# Перевод токенов
receiver = w3.eth.accounts[1]
transfer_amount = 1000

tx_hash = token.functions.transfer(receiver, transfer_amount).transact()
tx_receipt = w3.eth.wait_for_transaction_receipt(tx_hash)

# Проверка баланса после перевода
owner_balance = token.functions.balanceOf(deployer_account).call()
receiver_balance = token.functions.balanceOf(receiver).call()

print(f"Баланс владельца после перевода: {owner_balance} STK")
print(f"Баланс получателя после перевода: {receiver_balance} STK")

import hashlib
import json
import time
from typing import List, Dict, Any

class Block:
def __init__(self, index: int, transactions: List[Dict[str, Any]], timestamp: float, previous_hash: str):
self.index = index
self.transactions = transactions
self.timestamp = timestamp
self.previous_hash = previous_hash
self.nonce = 0
self.hash = self.calculate_hash()

def calculate_hash(self) -> str:
"""Вычисляет хеш блока."""
block_string = json.dumps({
"index": self.index,
"transactions": self.transactions,
"timestamp": self.timestamp,
"previous_hash": self.previous_hash,
"nonce": self.nonce
}, sort_keys=True).encode()

return hashlib.sha256(block_string).hexdigest()

def mine_block(self, difficulty: int) -> None:
"""Майнинг блока: поиск хеша, начинающегося с заданного числа нулей."""
target = "0" * difficulty

while self.hash[:difficulty] != target:
self.nonce += 1
self.hash = self.calculate_hash()

print(f"Block mined: {self.hash}")

class Blockchain:
def __init__(self, difficulty: int = 4):
self.chain: List[Block] = []
self.pending_transactions: List[Dict[str, Any]] = []
self.difficulty = difficulty
self.mining_reward = 100

# Создание генезис-блока
self.create_genesis_block()

def create_genesis_block(self) -> None:
"""Создает первый блок в цепи."""
genesis_block = Block(0, [], time.time(), "0")
genesis_block.hash = genesis_block.calculate_hash()
self.chain.append(genesis_block)

def get_latest_block(self) -> Block:
"""Возвращает последний блок в цепи."""
return self.chain[-1]

def mine_pending_transactions(self, miner_address: str) -> None:
"""Создает новый блок из ожидающих транзакций и добавляет его в цепь."""
# Добавление вознаграждения майнеру
self.pending_transactions.append({
"sender": "system",
"recipient": miner_address,
"amount": self.mining_reward
})

# Создание нового блока
block = Block(
len(self.chain),
self.pending_transactions,
time.time(),
self.get_latest_block().hash
)

# Майнинг блока
block.mine_block(self.difficulty)

# Добавление блока в цепь
self.chain.append(block)

# Очистка ожидающих транзакций
self.pending_transactions = []

def create_transaction(self, sender: str, recipient: str, amount: float) -> None:
"""Добавляет транзакцию в список ожидающих."""
self.pending_transactions.append({
"sender": sender,
"recipient": recipient,
"amount": amount
})

def get_balance(self, address: str) -> float:
"""Вычисляет баланс адреса на основе истории транзакций."""
balance = 0

for block in self.chain:
for transaction in block.transactions:
if transaction["sender"] == address:
balance -= transaction["amount"]
if transaction["recipient"] == address:
balance += transaction["amount"]

return balance

def is_chain_valid(self) -> bool:
"""Проверяет целостность всей цепи блоков."""
for i in range(1, len(self.chain)):
current_block = self.chain[i]
previous_block = self.chain[i – 1]

# Проверка хеша текущего блока
if current_block.hash != current_block.calculate_hash():
return False

# Проверка ссылки на хеш предыдущего блока
if current_block.previous_hash != previous_block.hash:
return False

return True

# Пример использования
if __name__ == "__main__":
# Создание блокчейна
my_blockchain = Blockchain(difficulty=4)

print("Майнинг первого блока...")
my_blockchain.mine_pending_transactions("miner-address-1")

# Создание нескольких транзакций
my_blockchain.create_transaction("address-1", "address-2", 50)
my_blockchain.create_transaction("address-2", "address-1", 20)

print("Майнинг второго блока...")
my_blockchain.mine_pending_transactions("miner-address-1")

# Проверка баланса майнера
print(f"Баланс майнера: {my_blockchain.get_balance('miner-address-1')}")

# Проверка целостности цепи
print(f"Блокчейн валиден: {my_blockchain.is_chain_valid()}")

# Вывод информации о цепи
for block in my_blockchain.chain:
print(f"Блок #{block.index} | Хеш: {block.hash} | Транзакций: {len(block.transactions)}")