Job Finance¶

job_finance — G6 Finance job agent.

Cluster: Job Agents | Type: component | MCP Tools: 26

Overview¶

Domain-specialist job agent for finance professionals including investment analysts, corporate finance teams, and risk managers. Analyses investment portfolios, forecasts market trends, assesses financial risk, generates regulatory reports, audits compliance, and performs competitive financial benchmarking — all within G6's safety-bounded, audit-trailed JobAgentBlock framework.

When to use:

Analysing an investment portfolio's risk/return profile against a benchmark index
Forecasting revenue and cost trends from historical financial data
Assessing regulatory compliance (Basel III, APRA, MiFID II) across financial instruments
Generating structured board-level financial reports from raw ledger data

Example:

from mvp.job_finance import JobFinanceBlock, JobFinanceInput

block = JobFinanceBlock()
result = block.infer(JobFinanceInput(
    task="Analyse the equity portfolio's VaR exposure at 95% confidence and recommend rebalancing actions",
    context={"portfolio_id": "EQ-APAC-01", "period": "2026-Q1", "benchmark": "ASX200"},
))
# result.ok → True; result.value → JobFinanceOutput with result, artifacts

Works well with: job_framework, job_accountant, job_analyst

Enterprise And Regulated-Use Caveat¶

Enterprise-oriented, not enterprise-certified

job_finance is suitable for MVP, internal analysis, and design-partner workflows where a qualified finance professional reviews the result. It is not, by itself, enterprise-grade finance software, investment advice, a trading system, or regulated compliance certification.

The component now includes enterprise-oriented controls: advisory MCP operations require structured professional attestation, live-data-dependent analysis can be run with enterprise_mode or require_live_data so it fails closed instead of silently using mock/fallback data, outputs include data_quality metadata, and enterprise-mode investment evaluation suppresses direct BUY/SELL/HOLD actions behind REVIEW_REQUIRED.

Before using this component in a production financial institution or regulated customer workflow, pair it with licensed market-data providers, authenticated reviewer workflows, immutable audit logs, RBAC, retention and privacy controls, model-validation evidence, jurisdiction-specific rule packs, and external legal/compliance review. Treat built-in SEC/MiFID/Dodd-Frank checks as heuristic decision support, not authoritative regulatory advice.

Enterprise controls exposed by this component:

Set parameters.enterprise_mode=True or parameters.require_live_data=True for live-data-dependent analysis that should fail closed when only sample/fallback data is available.
Inspect metadata.data_quality on MCP responses before using results in reports, approvals, or downstream automation.
Use structured attestation in context.attestation for advisory operations; a bare financial_advisor_context=True flag is intentionally insufficient.
Treat evaluate_investment outputs in enterprise mode as review artifacts. The component returns REVIEW_REQUIRED rather than executable trade advice.

Public API¶

`JobFinanceBlock(JobAgentBlock)`¶

G6 Finance job agent - delegates to finance MCP block.

Field	Type	Default
`name`	`str`	'job_finance'
`sector`	`SectorClassification`	field(default_factory=lambda: _SECTOR)
`toolkit`	`ToolkitSpec \\| None`	field(default_factory=lambda: JOB_TOOLKITS.get('finance'))
`mcp_module`	`str`	'mvp.job_finance.finance_mcp.server'
`agentic_planner`	`object \\| None`	None
`capabilities`	`ClassVar[set[type]]`	{Extensible, HumanLearnable, Collaborative, ProblemSolvable, KnowledgeGrounded, Memorable, AgentCommunicable, ExternallyAdaptable}

`MarketDataAdapter(ABC)`¶

Abstract market data source.

Methods:

`get_price(symbol: str) -> float | None`¶

Get current price for a symbol.

`get_history(symbol: str, days: int = 252) -> list[dict]`¶

Get historical OHLCV data. Returns list of dicts with

`get_fundamentals(symbol: str) -> dict | None`¶

Get fundamental data (P/E, EPS, market cap, etc.).

`get_filing(symbol: str, form_type: str) -> dict | None`¶

Get SEC filing data. form_type: '10-K', '10-Q', '8-K', etc.

`InMemoryMarketAdapter(MarketDataAdapter)`¶

In-memory market data adapter with sample stock data.

Methods:

`get_price(symbol: str) -> float | None`¶

`get_history(symbol: str, days: int = 252) -> list[dict]`¶

`get_fundamentals(symbol: str) -> dict | None`¶

`get_filing(symbol: str, form_type: str) -> dict | None`¶

`add_stock(symbol: str, data: dict) -> None`¶

Add or update a stock in the in-memory store.

`MockMarketAdapter(InMemoryMarketAdapter)`¶

Deterministic test adapter. Identical to InMemory but explicitly named.

Methods:

`get_filing(symbol: str, form_type: str) -> dict | None`¶

`YFinanceConnector(MarketDataAdapter)`¶

Real market data from Yahoo Finance via yfinance (free, no key needed).

Methods:

`data_quality() -> dict[str, Any]`¶

Return a self-describing data-quality tag for the most recent fetch.

`get_price(symbol: str) -> float | None`¶

`get_history(symbol: str, days: int = 252) -> list[dict]`¶

`get_fundamentals(symbol: str) -> dict | None`¶

`get_filing(symbol: str, form_type: str) -> dict | None`¶

Retrieve recent SEC filing via EDGAR (public, no key required).

`PortfolioRecord(BaseModel)`¶

A portfolio summary.

Field	Type	Default
`portfolio_id`	`str`	''
`name`	`str`	''
`strategy`	`str`	''
`benchmark`	`str`	''
`inception_date`	`str`	''
`total_value`	`float`	0.0
`positions_count`	`int`	0
`data`	`dict[str, Any]`	Field(default_factory=dict)

`PositionRecord(BaseModel)`¶

A single position.

Field	Type	Default
`position_id`	`str`	''
`portfolio_id`	`str`	''
`symbol`	`str`	''
`asset_class`	`str`	'equity'
`quantity`	`float`	0.0
`cost_basis`	`float`	0.0
`current_price`	`float`	0.0
`market_value`	`float`	0.0
`unrealized_pnl`	`float`	0.0
`weight`	`float`	0.0

`RiskMetrics(BaseModel)`¶

Risk assessment output.

Field	Type	Default
`var_95`	`float`	0.0
`var_99`	`float`	0.0
`cvar_95`	`float`	0.0
`sharpe_ratio`	`float`	0.0
`sortino_ratio`	`float`	0.0
`max_drawdown`	`float`	0.0
`beta`	`float`	1.0
`tracking_error`	`float`	0.0

`ValuationResult(BaseModel)`¶

Valuation analysis output.

Field	Type	Default
`method`	`str`	''
`intrinsic_value`	`float`	0.0
`current_price`	`float`	0.0
`upside_pct`	`float`	0.0
`pe_ratio`	`float`	0.0
`pb_ratio`	`float`	0.0
`ev_ebitda`	`float`	0.0
`dividend_yield`	`float`	0.0
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None

`ComplianceFlag(BaseModel)`¶

A single compliance finding.

Field	Type	Default
`rule`	`str`	''
`severity`	`str`	'info'
`symbol`	`str`	''
`message`	`str`	''

`JobFinanceInput(JobInput)`¶

Input for the Finance job agent.

`JobFinanceOutput(JobOutput)`¶

Output from the Finance job agent.

`JobFinanceMCPBlock(AIBlock[MCPJobFinanceInput, MCPJobFinanceOutput, dict])`¶

26-op MCP block for the Finance job agent.

Field	Type	Default
`name`	`str`	'job_finance_mcp'
`state`	`dict`	field(default_factory=dict)
`db_path`	`str`	':memory:'
`resource_bounds`	`ResourceBounds`	field(default_factory=ResourceBounds)
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)

Methods:

`infer(data: MCPJobFinanceInput) -> Result[MCPJobFinanceOutput]`¶

`MCPJobFinanceInput(BaseModel)`¶

Input to JobFinanceMCPBlock - 26-op dispatch.

Field	Type	Default
`op`	`Literal['analyze_portfolio', 'forecast_trend', 'assess_risk', 'generate_report', 'audit_compliance', 'analyze_market', 'evaluate_investment', 'assess_exposure', 'benchmark_returns', 'audit_transactions', 'create_proposal', 'review_deliverable', 'delegate_task', 'report_status', 'request_feedback', 'store_artifact', 'retrieve_artifact', 'list_artifacts', 'search_artifacts', 'archive', 'plan_sprint', 'track_progress', 'reflect_on_outcome', 'list_patterns', 'get_capabilities', 'info']`	required
`task`	`str`	''
`context`	`dict[str, Any]`	Field(default_factory=dict)
`parameters`	`dict[str, Any]`	Field(default_factory=dict)
`artifact_id`	`str`	''
`query`	`str`	''

`MCPJobFinanceOutput(BaseModel)`¶

Output from JobFinanceMCPBlock.

Field	Type	Default
`op`	`str`	required
`result`	`str`	''
`artifacts`	`list[dict[str, Any]]`	Field(default_factory=list)
`records`	`list[dict[str, Any]]`	Field(default_factory=list)
`message`	`str`	''
`count`	`int`	0
`found`	`bool`	False
`metadata`	`dict[str, Any]`	Field(default_factory=dict)
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None

`FinanceStore(JobStore)`¶

SQLite store for the Finance job agent.

Constructor:

Parameter	Type	Default
`db_path`	`str`	':memory:'

Methods:

`create_portfolio(name: str, strategy: str = '', benchmark: str = '', inception_date: str = '', data: dict | None = None) -> str`¶

`get_portfolio(portfolio_id: str) -> dict | None`¶

`list_portfolios(status: str = '') -> list[dict]`¶

`add_position(portfolio_id: str, symbol: str, asset_class: str = 'equity', quantity: float = 0, cost_basis: float = 0, current_price: float = 0, data: dict | None = None) -> str`¶

`get_positions(portfolio_id: str) -> list[dict]`¶

`update_position_price(position_id: str, current_price: float) -> bool`¶

`log_transaction(portfolio_id: str, symbol: str, tx_type: str, quantity: float, price: float, fees: float = 0, executed_at: str = '', data: dict | None = None) -> str`¶

`get_transactions(portfolio_id: str = '', symbol: str = '', limit: int = 100) -> list[dict]`¶

`store_risk_assessment(portfolio_id: str, var_95: float = 0, var_99: float = 0, sharpe_ratio: float = 0, max_drawdown: float = 0, data: dict | None = None) -> str`¶

`get_risk_assessments(portfolio_id: str, limit: int = 10) -> list[dict]`¶

`insert_market_data(symbol: str, date: str, open_: float, high: float, low: float, close: float, volume: float = 0, data: dict | None = None) -> str`¶

`get_market_data(symbol: str, start_date: str = '', end_date: str = '', limit: int = 500) -> list[dict]`¶

`get_latest_price(symbol: str) -> float | None`¶

Functions¶

`assemble_review_text(output: Any) -> str`¶

Collect the reviewable free text from a JobFinanceOutput (duck-typed).

`assess_finance_output(output: Any, qa_block: Any | None = None, generate: Any | None = None) -> GroundedRunResult`¶

Run grounded four-valued QA over a finance output.

`get_market_connector(name: str = 'yfinance') -> MarketDataAdapter`¶

Return a MarketDataAdapter instance by name.

`register_market_connector(name: str, cls: type[MarketDataAdapter]) -> None`¶

Register a custom connector class for DI (e.g., Bloomberg, Refinitiv).

MCP Tools¶

Operation	Source
`analyze_portfolio`	finance_mcp
`forecast_trend`	finance_mcp
`assess_risk`	finance_mcp
`generate_report`	finance_mcp
`audit_compliance`	finance_mcp
`analyze_market`	finance_mcp
`evaluate_investment`	finance_mcp
`assess_exposure`	finance_mcp
`benchmark_returns`	finance_mcp
`audit_transactions`	finance_mcp
`create_proposal`	finance_mcp
`review_deliverable`	finance_mcp
`delegate_task`	finance_mcp
`report_status`	finance_mcp
`request_feedback`	finance_mcp
`store_artifact`	finance_mcp
`retrieve_artifact`	finance_mcp
`list_artifacts`	finance_mcp
`search_artifacts`	finance_mcp
`archive`	finance_mcp
`plan_sprint`	finance_mcp
`track_progress`	finance_mcp
`reflect_on_outcome`	finance_mcp
`list_patterns`	finance_mcp
`get_capabilities`	finance_mcp
`info`	finance_mcp

Job Finance¶

Overview¶

Enterprise And Regulated-Use Caveat¶

Public API¶

JobFinanceBlock(JobAgentBlock)¶

MarketDataAdapter(ABC)¶

get_price(symbol: str) -> float | None¶

get_history(symbol: str, days: int = 252) -> list[dict]¶

get_fundamentals(symbol: str) -> dict | None¶

get_filing(symbol: str, form_type: str) -> dict | None¶

InMemoryMarketAdapter(MarketDataAdapter)¶

get_price(symbol: str) -> float | None¶

get_history(symbol: str, days: int = 252) -> list[dict]¶

get_fundamentals(symbol: str) -> dict | None¶

get_filing(symbol: str, form_type: str) -> dict | None¶

add_stock(symbol: str, data: dict) -> None¶

MockMarketAdapter(InMemoryMarketAdapter)¶

get_filing(symbol: str, form_type: str) -> dict | None¶

YFinanceConnector(MarketDataAdapter)¶

data_quality() -> dict[str, Any]¶

get_price(symbol: str) -> float | None¶

get_history(symbol: str, days: int = 252) -> list[dict]¶

get_fundamentals(symbol: str) -> dict | None¶

get_filing(symbol: str, form_type: str) -> dict | None¶

PortfolioRecord(BaseModel)¶

PositionRecord(BaseModel)¶

RiskMetrics(BaseModel)¶

ValuationResult(BaseModel)¶

ComplianceFlag(BaseModel)¶

JobFinanceInput(JobInput)¶

JobFinanceOutput(JobOutput)¶

JobFinanceMCPBlock(AIBlock[MCPJobFinanceInput, MCPJobFinanceOutput, dict])¶

infer(data: MCPJobFinanceInput) -> Result[MCPJobFinanceOutput]¶

MCPJobFinanceInput(BaseModel)¶

MCPJobFinanceOutput(BaseModel)¶

FinanceStore(JobStore)¶

create_portfolio(name: str, strategy: str = '', benchmark: str = '', inception_date: str = '', data: dict | None = None) -> str¶

get_portfolio(portfolio_id: str) -> dict | None¶

list_portfolios(status: str = '') -> list[dict]¶

add_position(portfolio_id: str, symbol: str, asset_class: str = 'equity', quantity: float = 0, cost_basis: float = 0, current_price: float = 0, data: dict | None = None) -> str¶

get_positions(portfolio_id: str) -> list[dict]¶

update_position_price(position_id: str, current_price: float) -> bool¶

log_transaction(portfolio_id: str, symbol: str, tx_type: str, quantity: float, price: float, fees: float = 0, executed_at: str = '', data: dict | None = None) -> str¶

get_transactions(portfolio_id: str = '', symbol: str = '', limit: int = 100) -> list[dict]¶

store_risk_assessment(portfolio_id: str, var_95: float = 0, var_99: float = 0, sharpe_ratio: float = 0, max_drawdown: float = 0, data: dict | None = None) -> str¶

get_risk_assessments(portfolio_id: str, limit: int = 10) -> list[dict]¶

insert_market_data(symbol: str, date: str, open_: float, high: float, low: float, close: float, volume: float = 0, data: dict | None = None) -> str¶

get_market_data(symbol: str, start_date: str = '', end_date: str = '', limit: int = 500) -> list[dict]¶

get_latest_price(symbol: str) -> float | None¶

Functions¶

assemble_review_text(output: Any) -> str¶

assess_finance_output(output: Any, qa_block: Any | None = None, generate: Any | None = None) -> GroundedRunResult¶

get_market_connector(name: str = 'yfinance') -> MarketDataAdapter¶

register_market_connector(name: str, cls: type[MarketDataAdapter]) -> None¶

MCP Tools¶

`JobFinanceBlock(JobAgentBlock)`¶

`MarketDataAdapter(ABC)`¶

`get_price(symbol: str) -> float | None`¶

`get_history(symbol: str, days: int = 252) -> list[dict]`¶

`get_fundamentals(symbol: str) -> dict | None`¶

`get_filing(symbol: str, form_type: str) -> dict | None`¶

`InMemoryMarketAdapter(MarketDataAdapter)`¶

`get_price(symbol: str) -> float | None`¶

`get_history(symbol: str, days: int = 252) -> list[dict]`¶

`get_fundamentals(symbol: str) -> dict | None`¶

`get_filing(symbol: str, form_type: str) -> dict | None`¶

`add_stock(symbol: str, data: dict) -> None`¶

`MockMarketAdapter(InMemoryMarketAdapter)`¶

`get_filing(symbol: str, form_type: str) -> dict | None`¶

`YFinanceConnector(MarketDataAdapter)`¶

`data_quality() -> dict[str, Any]`¶

`get_price(symbol: str) -> float | None`¶

`get_history(symbol: str, days: int = 252) -> list[dict]`¶

`get_fundamentals(symbol: str) -> dict | None`¶

`get_filing(symbol: str, form_type: str) -> dict | None`¶

`PortfolioRecord(BaseModel)`¶

`PositionRecord(BaseModel)`¶

`RiskMetrics(BaseModel)`¶

`ValuationResult(BaseModel)`¶

`ComplianceFlag(BaseModel)`¶

`JobFinanceInput(JobInput)`¶

`JobFinanceOutput(JobOutput)`¶

`JobFinanceMCPBlock(AIBlock[MCPJobFinanceInput, MCPJobFinanceOutput, dict])`¶

`infer(data: MCPJobFinanceInput) -> Result[MCPJobFinanceOutput]`¶

`MCPJobFinanceInput(BaseModel)`¶

`MCPJobFinanceOutput(BaseModel)`¶

`FinanceStore(JobStore)`¶

`create_portfolio(name: str, strategy: str = '', benchmark: str = '', inception_date: str = '', data: dict | None = None) -> str`¶

`get_portfolio(portfolio_id: str) -> dict | None`¶

`list_portfolios(status: str = '') -> list[dict]`¶

`add_position(portfolio_id: str, symbol: str, asset_class: str = 'equity', quantity: float = 0, cost_basis: float = 0, current_price: float = 0, data: dict | None = None) -> str`¶

`get_positions(portfolio_id: str) -> list[dict]`¶

`update_position_price(position_id: str, current_price: float) -> bool`¶

`log_transaction(portfolio_id: str, symbol: str, tx_type: str, quantity: float, price: float, fees: float = 0, executed_at: str = '', data: dict | None = None) -> str`¶

`get_transactions(portfolio_id: str = '', symbol: str = '', limit: int = 100) -> list[dict]`¶

`store_risk_assessment(portfolio_id: str, var_95: float = 0, var_99: float = 0, sharpe_ratio: float = 0, max_drawdown: float = 0, data: dict | None = None) -> str`¶

`get_risk_assessments(portfolio_id: str, limit: int = 10) -> list[dict]`¶

`insert_market_data(symbol: str, date: str, open_: float, high: float, low: float, close: float, volume: float = 0, data: dict | None = None) -> str`¶

`get_market_data(symbol: str, start_date: str = '', end_date: str = '', limit: int = 500) -> list[dict]`¶

`get_latest_price(symbol: str) -> float | None`¶

`assemble_review_text(output: Any) -> str`¶

`assess_finance_output(output: Any, qa_block: Any | None = None, generate: Any | None = None) -> GroundedRunResult`¶

`get_market_connector(name: str = 'yfinance') -> MarketDataAdapter`¶

`register_market_connector(name: str, cls: type[MarketDataAdapter]) -> None`¶