Python SDK

from c0gni import Agent, Strategy, MarketData
import asyncio

class SimpleArbitrageStrategy(Strategy):
    def __init__(self):
        super().__init__()
        self.min_profit_threshold = 0.005  # 0.5% minimum profit
        
    async def on_market_data(self, data: MarketData):
        # Check for arbitrage opportunities across DEXs
        opportunities = await self.scan_arbitrage_opportunities(data)
        
        for opportunity in opportunities:
            if opportunity.profit_percentage > self.min_profit_threshold:
                return self.arbitrage_signal(
                    buy_exchange=opportunity.buy_dex,
                    sell_exchange=opportunity.sell_dex,
                    token=opportunity.token,
                    profit_estimate=opportunity.profit_percentage
                )
        
        return None

# Create and configure agent
async def main():
    strategy = SimpleArbitrageStrategy()
    
    agent = Agent(
        name="My First Agent",
        strategy=strategy,
        initial_balance=1.0,  # 1 ETH
        network="sepolia"  # Start with testnet
    )
    
    # Start paper trading
    await agent.start_paper_trading(duration_hours=24)
    
    # Get results
    results = await agent.get_performance_summary()
    print(f"Total trades: {results.total_trades}")
    print(f"Win rate: {results.win_rate:.1%}")
    print(f"Total return: {results.total_return:.1%}")

if __name__ == "__main__":
    asyncio.run(main())

from c0gni.ml import PredictiveModel, FeatureExtractor
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score
import numpy as np

class MLEnhancedStrategy(Strategy):
    def __init__(self):
        super().__init__()
        self.feature_extractor = FeatureExtractor([
            'price_momentum_5m',
            'volume_ratio_1h', 
            'rsi_14',
            'macd_histogram',
            'bollinger_position'
        ])
        
        # Load pre-trained model or create new one
        self.model = PredictiveModel.load('price_direction_model')
        if self.model is None:
            self.model = self.train_model()
            
    def train_model(self):
        """Train a model to predict price direction"""
        # Get historical data for training
        historical_data = self.get_training_data(days=90)
        
        # Extract features
        X = []
        y = []
        
        for i in range(len(historical_data) - 10):
            # Features: current market state
            features = self.feature_extractor.extract(
                historical_data[i:i+1]
            )
            X.append(features)
            
            # Target: price direction in next 5 minutes
            current_price = historical_data[i]['close']
            future_price = historical_data[i+10]['close']  # 5min later
            y.append(1 if future_price > current_price else 0)
        
        # Train classifier
        model = RandomForestClassifier(n_estimators=100)
        model.fit(X, y)
        
        # Validate model
        accuracy = accuracy_score(y_test, model.predict(X_test))
        print(f"Model accuracy: {accuracy:.1%}")
        
        return model
    
    async def on_market_data(self, data):
        # Extract current features
        features = self.feature_extractor.extract([data])
        
        # Predict price direction
        prediction = self.model.predict_proba([features])[0]
        bullish_probability = prediction[1]
        
        # Generate signal if confidence is high
        if bullish_probability > 0.7:
            return self.buy_signal(
                confidence=bullish_probability,
                reason=f"ML prediction: {bullish_probability:.1%} bullish"
            )
        elif bullish_probability < 0.3:
            return self.sell_signal(
                confidence=1 - bullish_probability,
                reason=f"ML prediction: {1-bullish_probability:.1%} bearish"
            )
        
        return None

import pytest
from c0gni.testing import StrategyTester, MockMarketData
from my_strategy import TechnicalAnalysisStrategy

class TestTechnicalAnalysisStrategy:
    def setup_method(self):
        self.strategy = TechnicalAnalysisStrategy()
        self.tester = StrategyTester(self.strategy)
    
    def test_buy_signal_generation(self):
        # Create mock data with SMA crossover pattern
        mock_data = MockMarketData.create_sma_crossover(
            periods=50,
            crossover_at=30,
            price_start=100.0
        )
        
        # Feed data to strategy
        signals = []
        for data_point in mock_data:
            signal = await self.strategy.on_market_data(data_point)
            if signal:
                signals.append(signal)
        
        # Verify buy signal generated at crossover
        assert len(signals) == 1
        assert signals[0].action == "buy"
        assert signals[0].confidence > 0.7
    
    @pytest.mark.asyncio
    async def test_backtest_performance(self):
        # Run backtest on historical data
        results = await self.tester.backtest(
            start_date="2024-01-01",
            end_date="2024-01-31",
            initial_balance=2.0
        )
        
        # Performance assertions
        assert results.total_return > 0
        assert results.win_rate > 0.5
        assert results.max_drawdown < 0.3
    
    def test_risk_management(self):
        # Test position sizing logic
        balance = 10.0
        signal_confidence = 0.8
        
        position_size = self.strategy.calculate_position_size(
            balance, signal_confidence
        )
        
        # Should not risk more than 5% per trade
        assert position_size <= balance * 0.05