Spaces:
Running
Running
File size: 5,811 Bytes
da0ce7b | 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 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 | #!/usr/bin/env python3
"""
Audit the Aureon Python runtime graph.
This does not try to force every Python file into one execution graph. Instead,
it separates the repo into:
1. core runtime modules used by the live trading spine
2. standalone scripts/tools with no inbound links
3. internal imports that resolve to missing repo modules
"""
from __future__ import annotations
import ast
from collections import Counter, defaultdict
from pathlib import Path
from typing import Dict, Iterable, List, Set, Tuple
ROOT = Path(__file__).resolve().parents[2]
PACKAGE = ROOT / "aureon"
FOCUS_DIRS = {
"analytics",
"bridges",
"core",
"data_feeds",
"decoders",
"exchanges",
"harmonic",
"intelligence",
"monitors",
"portfolio",
"scanners",
"strategies",
"trading",
"utils",
}
RUNTIME_ROOTS = {
"aureon.exchanges.unified_market_trader",
"aureon.exchanges.capital_swarm_runner",
"aureon.exchanges.kraken_margin_penny_trader",
"aureon.exchanges.capital_cfd_trader",
"aureon.exchanges.alpaca_capital_runner",
"aureon.exchanges.alpaca_capital_style_trader",
}
def iter_focus_files() -> List[Path]:
files: List[Path] = []
for path in PACKAGE.rglob("*.py"):
if "__pycache__" in path.parts:
continue
rel = path.relative_to(PACKAGE)
if len(rel.parts) < 2:
continue
if rel.parts[0] in FOCUS_DIRS:
files.append(path)
return files
def module_name(path: Path) -> str:
return ".".join(path.relative_to(ROOT).with_suffix("").parts)
def resolve_import(current_mod: str, node: ast.ImportFrom) -> str:
if not node.module:
return ""
if node.level and current_mod.startswith("aureon."):
base = current_mod.split(".")[:-1]
up = max(0, len(base) - (node.level - 1))
return ".".join(base[:up] + node.module.split("."))
return node.module
def nearest_known_module(name: str, known: Set[str]) -> str:
parts = name.split(".")
for i in range(len(parts), 1, -1):
candidate = ".".join(parts[:i])
if candidate in known:
return candidate
return ""
def build_graph(paths: Iterable[Path]) -> Tuple[Dict[str, Set[str]], Counter, Dict[str, Set[str]], List[str], List[str]]:
known = {module_name(path) for path in paths}
graph: Dict[str, Set[str]] = defaultdict(set)
inbound: Counter = Counter()
missing: Dict[str, Set[str]] = defaultdict(set)
syntax_errors: List[str] = []
main_files: List[str] = []
for path in paths:
mod = module_name(path)
try:
tree = ast.parse(path.read_text(encoding="utf-8", errors="replace"))
except Exception as exc:
syntax_errors.append(f"{path.relative_to(ROOT)} :: {exc}")
continue
has_main = False
for node in ast.walk(tree):
if isinstance(node, ast.If):
try:
expr = ast.unparse(node.test)
except Exception:
expr = ""
if "__name__" in expr and "__main__" in expr:
has_main = True
elif isinstance(node, ast.Import):
for alias in node.names:
if not alias.name.startswith("aureon."):
continue
target = nearest_known_module(alias.name, known)
if target:
graph[mod].add(target)
inbound[target] += 1
elif isinstance(node, ast.ImportFrom):
resolved = resolve_import(mod, node)
if not resolved.startswith("aureon."):
continue
target = nearest_known_module(resolved, known)
if target:
graph[mod].add(target)
inbound[target] += 1
else:
missing[resolved].add(str(path.relative_to(ROOT)))
if has_main:
main_files.append(str(path.relative_to(ROOT)))
return graph, inbound, missing, syntax_errors, main_files
def reachable_from(roots: Iterable[str], graph: Dict[str, Set[str]]) -> Set[str]:
seen: Set[str] = set()
stack = [root for root in roots if root in graph or root in RUNTIME_ROOTS]
while stack:
node = stack.pop()
if node in seen:
continue
seen.add(node)
for dep in graph.get(node, set()):
if dep not in seen:
stack.append(dep)
return seen
def main() -> None:
files = iter_focus_files()
graph, inbound, missing, syntax_errors, main_files = build_graph(files)
known = sorted(module_name(path) for path in files)
reachable = reachable_from(RUNTIME_ROOTS, graph)
standalone = sorted(mod for mod in known if inbound.get(mod, 0) == 0 and mod not in RUNTIME_ROOTS)
print(f"FOCUS_MODULES {len(known)}")
print(f"SYNTAX_ERRORS {len(syntax_errors)}")
for item in syntax_errors[:20]:
print(f"SYNTAX {item}")
print(f"RUNTIME_REACHABLE {len(reachable)}")
for mod in sorted(reachable)[:80]:
print(f"REACHABLE {mod}")
print(f"STANDALONE_FOCUS {len(standalone)}")
for mod in standalone[:120]:
print(f"STANDALONE {mod}")
print(f"MAIN_FILES {len(main_files)}")
for mod in main_files[:80]:
print(f"MAIN {mod}")
print(f"MISSING_INTERNAL {len(missing)}")
for name, users in sorted(missing.items())[:80]:
joined = ", ".join(sorted(users)[:4])
print(f"MISSING {name} USED_BY {joined}")
if __name__ == "__main__":
main()
|