Files
playbook/test/test_tsl_api_framework.py
T

625 lines
22 KiB
Python

import csv
import json
import re
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
SKILL_ROOT = ROOT / "skills" / "tsl-api-reference"
FRAMEWORK_LOOKUP = SKILL_ROOT / "scripts" / "framework_lookup.py"
CLASS_LOOKUP = SKILL_ROOT / "scripts" / "class_lookup.py"
API_LOOKUP = SKILL_ROOT / "scripts" / "lookup.py"
FRAMEWORK_INDEX = SKILL_ROOT / "data" / "framework_index.json"
FUNCTION_INDEX = SKILL_ROOT / "data" / "function_index.tsv"
CODEGEN_ROOT = SKILL_ROOT / "references" / "codegen"
def run_script(script, *args):
return subprocess.run(
[sys.executable, str(script), *args],
cwd=ROOT,
capture_output=True,
text=True,
)
def indexed_class_count():
with FUNCTION_INDEX.open(encoding="utf-8", newline="") as handle:
return sum(
row["kind"] == "class"
for row in csv.DictReader(handle, delimiter="\t")
)
class TslApiFrameworkTests(unittest.TestCase):
def test_web_development_support_functions_are_removed(self):
removed = {
"Web_tb_arrow",
"Web_tb_bar",
"Web_tb_gradient",
"Web_tb_npbar",
"Web_tb_npgridbar",
"Web_css_bgbar",
"Web_css_bgcolor",
"Web_css_bghist",
"Web_richtext_zdbgcolor",
"Web_spark_hist",
"Web_spark_nphist",
"Web_fig_box",
"Web_fig_heatmap",
"Web_fig_histogram",
"Web_fig_radar",
"Web_fig_tree_decomp",
"Web_fig_violin",
"Demo_framestyle_constyle_dotweb",
"Demo_framestyle_constyle_net",
"Demo_framestyle_heatmaptb_net",
"Demo_framestyle_net",
"Demo_framestyle_web",
"Demo_framestyle_data",
"Kerneldensityestimator",
"Web_color_theme",
"Web_color_triscale",
"Web_dict2tabletree",
"Web_format",
"Web_getcolorscalebyrgbarr",
"Web_getdefultcolor",
"Web_getlineargradient",
"Web_gettdbar",
"Web_html_span",
"Web_html_table",
"Web_table2dicttree",
"web_metric_TranValuesWithGroup",
"tsplot_PerformanceSummary",
"tssubplot_lines",
}
with FUNCTION_INDEX.open(encoding="utf-8", newline="") as handle:
indexed = {
row["name"].casefold()
for row in csv.DictReader(handle, delimiter="\t")
}
headings = set()
for path in CODEGEN_ROOT.rglob("*.md"):
text = path.read_text(encoding="utf-8")
headings.update(
match.group(1).split("(", 1)[0].casefold()
for match in re.finditer(
r"^## `([^`]+)`", text, re.MULTILINE
)
)
self.assertTrue({name.casefold() for name in removed}.isdisjoint(indexed))
self.assertTrue({name.casefold() for name in removed}.isdisjoint(headings))
self.assertTrue(
{"barline_web", "fastline_web", "pieline_web"} <= indexed
)
def test_recent_update_source_names_do_not_leak_into_api_taxonomy(self):
fallback_pages = [
path.relative_to(CODEGEN_ROOT).as_posix()
for path in CODEGEN_ROOT.rglob("recent-updates-*.md")
]
with FUNCTION_INDEX.open(encoding="utf-8", newline="") as handle:
fallback_rows = [
row["page"]
for row in csv.DictReader(handle, delimiter="\t")
if "recent-updates-" in row["page"]
or row["module"].startswith("recent-updates-")
]
self.assertEqual([], fallback_pages)
self.assertEqual([], fallback_rows)
def test_class_index_covers_all_indexed_class_kinds(self):
result = run_script(CLASS_LOOKUP, "--list", "--format", "json")
self.assertEqual(0, result.returncode, result.stderr)
packet = json.loads(result.stdout)
self.assertEqual(indexed_class_count(), packet["count"])
classes = {
(item["scope"], item["qualified_name"]): item
for item in packet["classes"]
}
self.assertIn(("builtin", "TStringList"), classes)
self.assertIn(("module", "tsBackTesting"), classes)
self.assertIn(("module", "opDeltaHedging"), classes)
self.assertIn(("module", "tsLargeAndSmallTurning"), classes)
self.assertIn(("module", "regime_Based"), classes)
self.assertIn(("module", "ml_Clock"), classes)
self.assertIn(("module", "gaussianNB"), classes)
self.assertIn(("module", "decisionTreeClassifier"), classes)
self.assertIn(("module", "mlpClassifier"), classes)
self.assertIn(("module", "svc"), classes)
self.assertIn(("module", "brinson_PerforAttri"), classes)
self.assertIn(("module", "cb_BinaryTree"), classes)
self.assertIn(("module", "ts_ChipDistribution"), classes)
self.assertIn(("module", "tsPositionMeasure"), classes)
self.assertIn(("module", "ts_SameTypeFundEquityPercent"), classes)
self.assertIn(("module", "ts_AIPBackTesting"), classes)
self.assertIn(("dotnet", "tsOptimizer"), classes)
self.assertTrue(classes[("module", "tsBackTesting")]["profile_available"])
self.assertTrue(
classes[("module", "ts_SameTypeFundEquityPercent")][
"profile_available"
]
)
self.assertTrue(
classes[("module", "ts_AIPBackTesting")]["profile_available"]
)
self.assertFalse(classes[("builtin", "TStringList")]["profile_available"])
def test_recovered_recent_update_classes_keep_member_ownership(self):
with FUNCTION_INDEX.open(encoding="utf-8", newline="") as handle:
rows = list(csv.DictReader(handle, delimiter="\t"))
expected_owners = {
"module/derivatives-analysis/option-delta-hedging.md": {
"opDeltaHedging"
},
"module/tinysoft-strategies/market-cap-rotation.md": {
"tsLargeAndSmallTurning"
},
"module/portfolio-optimization/investment-clock.md": {
"regime_Based",
"ml_Clock",
},
}
for page, owners in expected_owners.items():
with self.subTest(page=page):
members = [
row
for row in rows
if row["page"] == page and row["kind"] != "class"
]
self.assertTrue(members)
self.assertTrue(all(row["owner"] in owners for row in members))
flattened_names = {
"GetOptionDelta",
"GetRegularAlpha",
"Anova",
"GetState_Division",
}
self.assertFalse(
any(
row["page"] in expected_owners
and row["name"] in flattened_names
and not row["owner"]
for row in rows
)
)
def test_recovered_domain_classes_and_wrappers_keep_their_boundaries(self):
with FUNCTION_INDEX.open(encoding="utf-8", newline="") as handle:
rows = list(csv.DictReader(handle, delimiter="\t"))
owned = {
"brinson_PerforAttri.GetBenchmark": "brinson_PerforAttri",
"cb_BinaryTree.CBValuation": "cb_BinaryTree",
"ts_ChipDistribution.ChipMeasure": "ts_ChipDistribution",
"tsPositionMeasure.PortfolioOptimize": "tsPositionMeasure",
"ts_SameTypeFundEquityPercent.GetFundArr": (
"ts_SameTypeFundEquityPercent"
),
"ts_AIPBackTesting.CumReturn": "ts_AIPBackTesting",
}
for qualified_name, owner in owned.items():
with self.subTest(qualified_name=qualified_name):
matches = [
row for row in rows if row["qualified_name"] == qualified_name
]
self.assertTrue(matches)
self.assertTrue(all(row["owner"] == owner for row in matches))
wrappers = {
"brinson_perforAttri_demo",
"op_underlying_hv",
"Index_TA_ChipDistribution",
"CB_CCB",
"CB_GetRiskTarget_CCB",
"getstockVal",
}
for name in wrappers:
with self.subTest(name=name):
matches = [row for row in rows if row["name"] == name]
self.assertTrue(matches)
self.assertTrue(all(not row["owner"] for row in matches))
flattened_members = {
"GetBenchmark",
"CBValuation",
"ChipMeasure",
"PortfolioOptimize",
"GetFundArr",
"CumReturn",
}
self.assertFalse(
any(
row["scope"] == "module"
and row["name"] in flattened_members
and row["page"].startswith("module/")
and not row["owner"]
for row in rows
)
)
def test_ts_ops_indexes_runtime_class_without_example_implementation(self):
with FUNCTION_INDEX.open(encoding="utf-8", newline="") as handle:
rows = list(csv.DictReader(handle, delimiter="\t"))
page_rows = [
row for row in rows if row["page"] == "module/ts-ops.md"
]
self.assertIn(
("opsServer", "class", ""),
{
(row["qualified_name"], row["kind"], row["owner"])
for row in page_rows
},
)
expected_members = {
"opsServer.FOpsId",
"opsServer.FAsyncResults",
"opsServer.FLastId",
"opsServer.FDataUrl",
"opsServer.FOPSList",
"opsServer.FOPSName",
"opsServer.FCallBack",
"opsServer.FSubCallBack",
"opsServer.FReturnRuntimeError",
"opsServer.FDebug",
"opsServer.create",
"opsServer.CallBack",
"opsServer.WaitSubscription",
"opsServer.AsyncCall",
"opsServer.DoOnMessage",
}
self.assertTrue(
expected_members
<= {row["qualified_name"] for row in page_rows}
)
self.assertTrue(
all(
row["owner"] == "opsServer"
for row in page_rows
if row["kind"] != "class"
)
)
example_names = {
"myopsserver2",
"hahaha2",
"getmydata",
"my_multiply",
}
self.assertFalse(
any(row["name"] in example_names for row in page_rows)
)
ops_call = [
row
for row in rows
if row["name"] == "opsCall"
and row["page"] == "module/ts-opi.md"
]
self.assertTrue(ops_call)
self.assertTrue(
all(row["kind"] == "function" and not row["owner"] for row in ops_call)
)
def test_recovered_machine_learning_classes_have_core_members(self):
expected = {
"gaussianNB": {"create", "Fit", "Predict", "Predict_proba"},
"decisionTreeClassifier": {
"create",
"Fit",
"Predict",
"Predict_proba",
},
"mlpClassifier": {"create", "Fit", "Predict", "Score"},
"svc": {"create", "Fit", "Predict", "Predict_proba", "Score"},
}
for class_name, required_members in expected.items():
with self.subTest(class_name=class_name):
result = run_script(
CLASS_LOOKUP,
"--class",
class_name,
"--include-members",
"--format",
"json",
)
self.assertEqual(0, result.returncode, result.stderr)
members = {
item["name"] for item in json.loads(result.stdout)["members"]
}
self.assertTrue(required_members <= members)
def test_generic_class_uses_compact_member_view_without_fake_lifecycle(self):
result = run_script(
CLASS_LOOKUP,
"--class",
"TStringList",
"--format",
"json",
)
self.assertEqual(0, result.returncode, result.stderr)
packet = json.loads(result.stdout)
self.assertEqual("resolved", packet["status"])
self.assertEqual("not_profiled", packet["profile_status"])
self.assertIsNone(packet["profile"])
self.assertEqual({"method": 25, "property": 15}, packet["member_summary"])
def test_profiled_class_adds_framework_lifecycle(self):
result = run_script(
CLASS_LOOKUP,
"--class",
"tsBackTesting",
"--format",
"json",
)
self.assertEqual(0, result.returncode, result.stderr)
packet = json.loads(result.stdout)
self.assertEqual("resolved", packet["profile_status"])
self.assertEqual("resolved", packet["profile"]["scaffold_status"])
self.assertEqual("incomplete", packet["profile"]["contract_status"])
self.assertEqual(
["construct", "configure", "schedule", "callback", "execute", "inspect"],
[phase["phase"] for phase in packet["profile"]["lifecycle"]],
)
def test_non_backtest_framework_archetypes_have_profiles(self):
expected = {
"tsbf_ReturnAttribution": {
"phases": ["construct", "configure", "schedule", "callback", "inspect"],
"contract_status": "incomplete",
},
"ts_EvaluationOfRotationSkill": {
"phases": ["construct", "execute", "inspect"],
"contract_status": "resolved",
},
"ts_IndustryAllocationOfFund": {
"phases": ["construct", "configure", "execute"],
"contract_status": "resolved",
},
"ts_SameTypeFundEquityPercent": {
"phases": [
"construct",
"configure",
"schedule",
"callback",
"execute",
"inspect",
],
"contract_status": "incomplete",
},
"ts_AIPBackTesting": {
"phases": [
"construct",
"configure",
"schedule",
"callback",
"execute",
"inspect",
],
"contract_status": "incomplete",
},
}
for class_name, contract in expected.items():
with self.subTest(class_name=class_name):
result = run_script(
CLASS_LOOKUP,
"--class",
class_name,
"--format",
"json",
)
self.assertEqual(0, result.returncode, result.stderr)
profile = json.loads(result.stdout)["profile"]
self.assertEqual(
contract["phases"],
[phase["phase"] for phase in profile["lifecycle"]],
)
self.assertEqual(
contract["contract_status"], profile["contract_status"]
)
def test_tsoptimizer_resolves_after_module_migration(self):
unfiltered = run_script(
CLASS_LOOKUP,
"--class",
"tsOptimizer",
"--format",
"json",
)
filtered = run_script(
CLASS_LOOKUP,
"--class",
"tsOptimizer",
"--scope",
"dotnet",
"--module",
"math",
"--format",
"json",
)
self.assertEqual(0, unfiltered.returncode, unfiltered.stderr)
unfiltered_packet = json.loads(unfiltered.stdout)
self.assertEqual("resolved", unfiltered_packet["status"])
self.assertEqual("dotnet", unfiltered_packet["class"]["scope"])
self.assertEqual("math", unfiltered_packet["class"]["module"])
self.assertEqual(0, filtered.returncode, filtered.stderr)
self.assertEqual(
"dotnet",
json.loads(filtered.stdout)["class"]["scope"],
)
self.assertEqual(
"math",
json.loads(filtered.stdout)["class"]["module"],
)
def test_class_check_validates_generic_and_profile_layers(self):
result = run_script(CLASS_LOOKUP, "--check")
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn(f"OK: {indexed_class_count()} classes indexed", result.stdout)
self.assertIn("6 framework profile(s) validated", result.stdout)
def test_framework_index_references_existing_api_entries(self):
result = run_script(FRAMEWORK_LOOKUP, "--check")
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn("OK: 6 framework profile(s) validated", result.stdout)
def test_ts_backtesting_returns_class_first_scaffold(self):
result = run_script(
FRAMEWORK_LOOKUP,
"--framework",
"tsBackTesting",
"--format",
"json",
)
self.assertEqual(0, result.returncode, result.stderr)
packet = json.loads(result.stdout)
self.assertEqual("resolved", packet["scaffold_status"])
self.assertEqual("incomplete", packet["contract_status"])
self.assertEqual(
["construct", "configure", "schedule", "callback", "execute", "inspect"],
[phase["phase"] for phase in packet["lifecycle"]],
)
self.assertEqual(
["tsBackTesting.GetTradeOrder"],
[hook["api"] for hook in packet["hooks"] if hook["required"]],
)
self.assertEqual(
"CONTRACT_INCOMPLETE",
packet["diagnostics"][0]["code"],
)
def test_candidate_discriminator_does_not_invent_a_mode(self):
result = run_script(
FRAMEWORK_LOOKUP,
"--framework",
"tsBackTesting",
"--config",
"FGroupType=1",
"--format",
"json",
)
self.assertEqual(0, result.returncode, result.stderr)
packet = json.loads(result.stdout)
self.assertEqual({"FGroupType": "1"}, packet["configured_values"])
self.assertEqual("incomplete", packet["contract_status"])
self.assertIn(
"CONTRACT_MODE_UNRESOLVED",
{item["code"] for item in packet["diagnostics"]},
)
def test_unknown_configuration_is_reported(self):
result = run_script(
FRAMEWORK_LOOKUP,
"--framework",
"tsBackTesting",
"--config",
"notAField=1",
"--format",
"json",
)
self.assertEqual(0, result.returncode, result.stderr)
packet = json.loads(result.stdout)
self.assertIn(
"UNKNOWN_CONFIGURATION",
{item["code"] for item in packet["diagnostics"]},
)
def test_unknown_framework_uses_query_success_exit_code(self):
result = run_script(
FRAMEWORK_LOOKUP,
"--framework",
"MissingFramework",
"--format",
"json",
)
self.assertEqual(0, result.returncode, result.stderr)
self.assertEqual("not_found", json.loads(result.stdout)["status"])
def test_check_rejects_stale_evidence(self):
data = json.loads(FRAMEWORK_INDEX.read_text(encoding="utf-8"))
data["frameworks"][0]["hooks"][0]["evidence"] = [
"module/ts-backtesting.md#missing"
]
with tempfile.TemporaryDirectory() as temp_dir:
index_path = Path(temp_dir) / "framework_index.json"
index_path.write_text(json.dumps(data), encoding="utf-8")
result = run_script(
FRAMEWORK_LOOKUP,
"--check",
"--index",
str(index_path),
)
self.assertEqual(1, result.returncode)
self.assertIn("does not identify an indexed overload", result.stderr)
def test_query_reports_invalid_profile_as_data_error(self):
data = json.loads(FRAMEWORK_INDEX.read_text(encoding="utf-8"))
data["frameworks"][0]["execution"][0]["evidence"] = [
"module/ts-backtesting.md#missing"
]
with tempfile.TemporaryDirectory() as temp_dir:
index_path = Path(temp_dir) / "framework_index.json"
index_path.write_text(json.dumps(data), encoding="utf-8")
result = run_script(
FRAMEWORK_LOOKUP,
"--framework",
"tsBackTesting",
"--format",
"json",
"--index",
str(index_path),
)
self.assertEqual(1, result.returncode)
self.assertEqual("data_error", json.loads(result.stdout)["status"])
def test_check_rejects_profile_without_indexed_class(self):
data = json.loads(FRAMEWORK_INDEX.read_text(encoding="utf-8"))
data["frameworks"][0]["qualified_name"] = "MissingClass"
with tempfile.TemporaryDirectory() as temp_dir:
index_path = Path(temp_dir) / "framework_index.json"
index_path.write_text(json.dumps(data), encoding="utf-8")
result = run_script(
FRAMEWORK_LOOKUP,
"--check",
"--index",
str(index_path),
)
self.assertEqual(1, result.returncode)
self.assertIn("is not an indexed class", result.stderr)
def test_ordinary_api_lookup_remains_available(self):
result = run_script(API_LOOKUP, "--name", "tsBackTesting.BackTest")
self.assertEqual(0, result.returncode, result.stderr)
self.assertIn("### `BackTest()`", result.stdout)
self.assertIn("module/ts-backtesting.md#backtest", result.stdout)
if __name__ == "__main__":
unittest.main()