🎯 Initial commit: Uniswap Auto CLP trading system
Core Components: - uniswap_manager.py: V3 concentrated liquidity position manager - clp_hedger.py: Hyperliquid perpetuals hedging bot - requirements.txt: Python dependencies - .gitignore: Security exclusions for sensitive data - doc/: Project documentation - tools/: Utility scripts and Git agent Features: - Automated liquidity provision on Uniswap V3 (WETH/USDC) - Delta-neutral hedging using Hyperliquid perpetuals - Position lifecycle management (open/close/rebalance) - Automated backup and version control system Security: - Private keys and tokens excluded from version control - Environment variables properly handled - Automated security validation for backups Git Agent: - Hourly automated backups to separate branches - Keep last 100 backups (~4 days coverage) - Detailed change tracking and parameter monitoring - Push to Gitea server automatically - Manual main branch control preserved - No performance tracking for privacy - No notifications for simplicity Files Added: - git_agent.py: Main automation script - agent_config.json: Configuration with Gitea settings - git_utils.py: Git operations wrapper - backup_manager.py: Backup branch management - change_detector.py: File change analysis - cleanup_manager.py: 100-backup rotation - commit_formatter.py: Detailed commit messages - README_GIT_AGENT.md: Complete usage documentation
This commit is contained in:
153
tools/cleanup_manager.py
Normal file
153
tools/cleanup_manager.py
Normal file
@ -0,0 +1,153 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Cleanup Manager for Git Agent
|
||||
Manages backup branch rotation (keep last 100)
|
||||
"""
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import logging
|
||||
from typing import Dict, Any, List
|
||||
|
||||
class CleanupManager:
|
||||
"""Manages backup branch cleanup and rotation"""
|
||||
|
||||
def __init__(self, config: Dict[str, Any], logger: logging.Logger):
|
||||
self.config = config
|
||||
self.logger = logger
|
||||
self.backup_config = config.get('backup', {})
|
||||
self.prefix = self.backup_config.get('branch_prefix', 'backup-')
|
||||
self.max_backups = self.backup_config.get('keep_max_count', 100)
|
||||
self.project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
def cleanup_old_backups(self) -> bool:
|
||||
"""Clean up old backup branches to keep only the last N"""
|
||||
try:
|
||||
# Get all backup branches
|
||||
backup_branches = self._get_backup_branches()
|
||||
|
||||
if len(backup_branches) <= self.max_backups:
|
||||
self.logger.info(f"✅ Backup count ({len(backup_branches)}) within limit ({self.max_backups})")
|
||||
return False # No cleanup needed
|
||||
|
||||
# Branches to delete (oldest ones)
|
||||
branches_to_delete = backup_branches[self.max_backups:]
|
||||
|
||||
if not branches_to_delete:
|
||||
return False
|
||||
|
||||
self.logger.info(f"🧹 Cleaning up {len(branches_to_delete)} old backup branches")
|
||||
|
||||
deleted_count = 0
|
||||
for branch in branches_to_delete:
|
||||
# Delete local branch
|
||||
if self._delete_local_branch(branch):
|
||||
# Delete remote branch
|
||||
if self._delete_remote_branch(branch):
|
||||
deleted_count += 1
|
||||
self.logger.debug(f" ✅ Deleted: {branch}")
|
||||
else:
|
||||
self.logger.warning(f" ⚠️ Local deleted, remote failed: {branch}")
|
||||
else:
|
||||
self.logger.warning(f" ❌ Failed to delete: {branch}")
|
||||
|
||||
if deleted_count > 0:
|
||||
self.logger.info(f"✅ Cleanup completed: deleted {deleted_count} old backup branches")
|
||||
return True
|
||||
else:
|
||||
self.logger.warning("⚠️ No branches were successfully deleted")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"❌ Cleanup failed: {e}")
|
||||
return False
|
||||
|
||||
def _get_backup_branches(self) -> List[str]:
|
||||
"""Get all backup branches sorted by timestamp (newest first)"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
['git', 'branch', '-a'],
|
||||
cwd=self.project_root,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False
|
||||
)
|
||||
|
||||
if result.returncode != 0:
|
||||
return []
|
||||
|
||||
branches = []
|
||||
for line in result.stdout.strip().split('\n'):
|
||||
if line.strip():
|
||||
# Clean up branch name
|
||||
branch = line.strip().replace('* ', '').replace('remotes/origin/', '')
|
||||
if branch.startswith(self.prefix):
|
||||
branches.append(branch)
|
||||
|
||||
# Sort by timestamp (extract from branch name)
|
||||
# Format: backup-YYYY-MM-DD-HH
|
||||
branches.sort(key=lambda x: x.replace(self.prefix, ''), reverse=True)
|
||||
return branches
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Error getting backup branches: {e}")
|
||||
return []
|
||||
|
||||
def _delete_local_branch(self, branch_name: str) -> bool:
|
||||
"""Delete local branch"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
['git', 'branch', '-D', branch_name],
|
||||
cwd=self.project_root,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
return True
|
||||
else:
|
||||
self.logger.debug(f"Local delete failed for {branch_name}: {result.stderr}")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Exception deleting local branch {branch_name}: {e}")
|
||||
return False
|
||||
|
||||
def _delete_remote_branch(self, branch_name: str) -> bool:
|
||||
"""Delete remote branch"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
['git', 'push', 'origin', '--delete', branch_name],
|
||||
cwd=self.project_root,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
return True
|
||||
else:
|
||||
# Might already be deleted remotely, that's ok
|
||||
if "not found" in result.stderr.lower() or "does not exist" in result.stderr.lower():
|
||||
return True
|
||||
self.logger.debug(f"Remote delete failed for {branch_name}: {result.stderr}")
|
||||
return False
|
||||
|
||||
except Exception as e:
|
||||
self.logger.error(f"Exception deleting remote branch {branch_name}: {e}")
|
||||
return False
|
||||
|
||||
def get_cleanup_stats(self) -> Dict[str, Any]:
|
||||
"""Get statistics about backup cleanup"""
|
||||
backup_branches = self._get_backup_branches()
|
||||
current_count = len(backup_branches)
|
||||
|
||||
return {
|
||||
'current_backup_count': current_count,
|
||||
'max_allowed': self.max_backups,
|
||||
'cleanup_needed': current_count > self.max_backups,
|
||||
'branches_to_delete': max(0, current_count - self.max_backups),
|
||||
'newest_backup': backup_branches[0] if backup_branches else None,
|
||||
'oldest_backup': backup_branches[-1] if backup_branches else None
|
||||
}
|
||||
Reference in New Issue
Block a user