monsoon/run.py at main · kcolbchain/monsoon · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
#!/usr/bin/env python3
"""
kcolbchain monsoon runner.

Usage:
    python run.py                              # default config/default.yaml
    python run.py --config config/live.yaml    # alternate config
    python run.py --ticks 5 --simulate         # force simulation
"""
import argparse
import logging
import os
import yaml
from pathlib import Path

from src.agent.farmer import FarmingAgent
from src.agent.wallet_manager import WalletManager, Wallet
from src.strategies.bridge_strategy import BridgeStrategy
from src.strategies.dex_strategy import DexStrategy
from src.strategies.solana_strategy import SolanaStrategy


def load_config(path: str) -> dict:
    with open(path) as f:
        return yaml.safe_load(f)


def setup_wallets(config: dict) -> WalletManager:
    agent_cfg = config.get("agent", {})
    wm = WalletManager(simulate=agent_cfg.get("simulate", True))
    wallets_cfg = config.get("wallets", [])

    for i, w in enumerate(wallets_cfg):
        env_key = f"WALLET_{i+1}_KEY"
        pk = os.environ.get(env_key)
        if not pk:
            logging.warning(f"No private key for {w.get('label', f'wallet-{i}')} "
                          f"(set {env_key} env var). Skipping.")
            continue

        from eth_account import Account
        acct = Account.from_key(pk)
        wallet = Wallet(
            address=acct.address,
            label=w.get("label", f"wallet-{i}"),
            private_key=pk,
        )
        wm.add_wallet(wallet)
        logging.info(f"Loaded wallet: {wallet.label} ({wallet.address[:8]}...)")

    return wm


def setup_strategies(config: dict) -> list:
    strategies = []
    strat_cfg = config.get("strategies", {})

    if strat_cfg.get("bridge", {}).get("enabled", True):
        strategies.append(BridgeStrategy())

    if strat_cfg.get("dex", {}).get("enabled", True):
        strategies.append(DexStrategy())

    if strat_cfg.get("solana", {}).get("enabled", False):
        network = strat_cfg.get("solana", {}).get("network", "mainnet")
        simulate = config.get("agent", {}).get("simulate", True)
        strategies.append(SolanaStrategy(network=network, simulate=simulate))

    return strategies


def main():
    parser = argparse.ArgumentParser(description="kcolbchain airdrop farming agent")
    parser.add_argument("--config", default="config/default.yaml", help="Config file path")
    parser.add_argument("--ticks", type=int, default=10, help="Number of action ticks")
    parser.add_argument("--simulate", action="store_true", help="Force simulation mode")
    args = parser.parse_args()

    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)-8s [%(name)s] %(message)s",
        datefmt="%H:%M:%S",
    )

    config = load_config(args.config)
    if args.simulate:
        config.setdefault("agent", {})["simulate"] = True

    agent_cfg = config.get("agent", {})
    wm = setup_wallets(config)

    if len(wm.wallets) == 0:
        logging.error("No wallets loaded. Set WALLET_1_KEY env var.")
        logging.info("Running in simulate mode with dummy wallet...")
        agent_cfg["simulate"] = True
        dummy = Wallet(address="0x" + "0" * 40, label="simulate-dummy")
        wm.add_wallet(dummy)

    agent = FarmingAgent(wm, agent_cfg)

    for strategy in setup_strategies(config):
        agent.add_strategy(strategy)

    mode = "SIMULATE" if agent_cfg.get("simulate", True) else "LIVE"
    logging.info(f"=== kcolbchain monsoon [{mode}] ===")
    logging.info(f"Wallets: {len(wm.wallets)}, Strategies: {len(agent.strategies)}, Ticks: {args.ticks}")

    agent.run(ticks=args.ticks)


if __name__ == "__main__":
    main()