Initial commit: A-share stock analysis project with screening, backtesting, and multi-factor analysis tools
This commit is contained in:
@@ -0,0 +1,89 @@
|
||||
"""
|
||||
Strategy backtesting engine for A-share stocks.
|
||||
|
||||
Supports:
|
||||
- Predefined strategies (MA crossover, momentum, mean reversion, etc.)
|
||||
- Custom entry/exit rules
|
||||
- Multi-strategy comparison
|
||||
- Performance metrics (returns, drawdown, Sharpe, win rate)
|
||||
|
||||
Usage: python3 backtest_engine.py <function> <json_args>
|
||||
"""
|
||||
|
||||
import sys
|
||||
import json
|
||||
|
||||
|
||||
def run(strategy_name=None, entry_rule=None, exit_rule=None,
|
||||
universe="hs300", symbols=None, start_date="20210101", end_date="20251231"):
|
||||
"""Run a single backtest. Stub - to be implemented."""
|
||||
return json.dumps({
|
||||
"status": "stub",
|
||||
"message": f"Backtest (strategy={strategy_name or 'custom'}, universe={universe}) — not yet implemented",
|
||||
"metrics": {
|
||||
"cumulative_return": None,
|
||||
"annualized_return": None,
|
||||
"win_rate": None,
|
||||
"max_drawdown": None,
|
||||
"sharpe_ratio": None,
|
||||
"benchmark_return": None,
|
||||
},
|
||||
}, ensure_ascii=False)
|
||||
|
||||
|
||||
def predefined():
|
||||
"""List available predefined strategies."""
|
||||
strategies = [
|
||||
{
|
||||
"name": "ma_cross",
|
||||
"description": "MA Golden Cross: buy when 20-day MA crosses above 60-day MA",
|
||||
"params": {"fast": 20, "slow": 60},
|
||||
},
|
||||
{
|
||||
"name": "momentum_breakout",
|
||||
"description": "Momentum Breakout: buy when price breaks 20-day high with volume > 1.5x avg",
|
||||
"params": {"lookback": 20, "volume_multiplier": 1.5},
|
||||
},
|
||||
{
|
||||
"name": "mean_reversion",
|
||||
"description": "Mean Reversion: buy when price deviates >2 std below 20-day MA, sell at MA",
|
||||
"params": {"ma_period": 20, "std_dev": 2},
|
||||
},
|
||||
{
|
||||
"name": "turtle",
|
||||
"description": "Turtle Trading: breakout of 20-day high, exit at 10-day low",
|
||||
"params": {"entry_period": 20, "exit_period": 10},
|
||||
},
|
||||
]
|
||||
return json.dumps({"strategies": strategies}, ensure_ascii=False)
|
||||
|
||||
|
||||
def compare(strategies, universe="hs300", start_date="20210101", end_date="20251231"):
|
||||
"""Compare multiple strategies. Stub - to be implemented."""
|
||||
return json.dumps({
|
||||
"status": "stub",
|
||||
"message": f"Strategy comparison ({strategies}) — not yet implemented",
|
||||
"comparison": [],
|
||||
}, ensure_ascii=False)
|
||||
|
||||
|
||||
FUNCTIONS = {
|
||||
"run": run,
|
||||
"predefined": predefined,
|
||||
"compare": compare,
|
||||
}
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: backtest_engine.py <function> [json_args]")
|
||||
sys.exit(1)
|
||||
|
||||
func_name = sys.argv[1]
|
||||
args = json.loads(sys.argv[2]) if len(sys.argv) > 2 else {}
|
||||
|
||||
if func_name not in FUNCTIONS:
|
||||
print(f"Unknown function: {func_name}")
|
||||
sys.exit(1)
|
||||
|
||||
result = FUNCTIONS[func_name](**args)
|
||||
print(result)
|
||||
@@ -0,0 +1,81 @@
|
||||
"""
|
||||
A-share market data fetching module.
|
||||
Uses AKShare as the primary data source.
|
||||
|
||||
Usage: python3 market_data.py <function> <json_args>
|
||||
"""
|
||||
|
||||
import sys
|
||||
import json
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
|
||||
def get_quote(symbol, period="daily", start_date=None, end_date=None):
|
||||
"""Get K-line data for a stock. Stub - to be implemented with AKShare."""
|
||||
# TODO: Implement with akshare.stock_zh_a_hist()
|
||||
return json.dumps({
|
||||
"status": "stub",
|
||||
"message": f"Quote for {symbol} ({period}) — not yet implemented",
|
||||
"symbol": symbol,
|
||||
"period": period,
|
||||
}, ensure_ascii=False)
|
||||
|
||||
|
||||
def get_financial(symbol):
|
||||
"""Get financial indicators. Stub - to be implemented with AKShare."""
|
||||
# TODO: Implement with akshare.stock_financial_analysis_indicator()
|
||||
return json.dumps({
|
||||
"status": "stub",
|
||||
"message": f"Financial data for {symbol} — not yet implemented",
|
||||
}, ensure_ascii=False)
|
||||
|
||||
|
||||
def get_moneyflow(symbol, days=10):
|
||||
"""Get capital flow data. Stub - to be implemented with AKShare."""
|
||||
# TODO: Implement with akshare.stock_individual_fund_flow()
|
||||
return json.dumps({
|
||||
"status": "stub",
|
||||
"message": f"Money flow for {symbol} ({days}d) — not yet implemented",
|
||||
}, ensure_ascii=False)
|
||||
|
||||
|
||||
def get_index(index_code="all", days=30):
|
||||
"""Get index data. Stub - to be implemented with AKShare."""
|
||||
# TODO: Implement with akshare.stock_zh_index_daily()
|
||||
return json.dumps({
|
||||
"status": "stub",
|
||||
"message": f"Index data ({index_code}, {days}d) — not yet implemented",
|
||||
}, ensure_ascii=False)
|
||||
|
||||
|
||||
def get_sector(date=None):
|
||||
"""Get sector performance. Stub - to be implemented with AKShare."""
|
||||
# TODO: Implement with akshare.stock_board_industry_name_em()
|
||||
return json.dumps({
|
||||
"status": "stub",
|
||||
"message": "Sector data — not yet implemented",
|
||||
}, ensure_ascii=False)
|
||||
|
||||
|
||||
FUNCTIONS = {
|
||||
"quote": get_quote,
|
||||
"financial": get_financial,
|
||||
"moneyflow": get_moneyflow,
|
||||
"index": get_index,
|
||||
"sector": get_sector,
|
||||
}
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: market_data.py <function> [json_args]")
|
||||
sys.exit(1)
|
||||
|
||||
func_name = sys.argv[1]
|
||||
args = json.loads(sys.argv[2]) if len(sys.argv) > 2 else {}
|
||||
|
||||
if func_name not in FUNCTIONS:
|
||||
print(f"Unknown function: {func_name}")
|
||||
sys.exit(1)
|
||||
|
||||
result = FUNCTIONS[func_name](**args)
|
||||
print(result)
|
||||
@@ -0,0 +1,47 @@
|
||||
"""
|
||||
News and sentiment analysis for A-share stocks.
|
||||
|
||||
Usage: python3 sentiment.py <function> <json_args>
|
||||
"""
|
||||
|
||||
import sys
|
||||
import json
|
||||
|
||||
|
||||
def news(symbol, limit=20):
|
||||
"""Get recent news for a stock. Stub - to be implemented with AKShare."""
|
||||
return json.dumps({
|
||||
"status": "stub",
|
||||
"message": f"News for {symbol} — not yet implemented",
|
||||
"items": [],
|
||||
}, ensure_ascii=False)
|
||||
|
||||
|
||||
def market_sentiment():
|
||||
"""Get overall market sentiment score. Stub - to be implemented."""
|
||||
return json.dumps({
|
||||
"status": "stub",
|
||||
"message": "Market sentiment — not yet implemented",
|
||||
"score": None,
|
||||
}, ensure_ascii=False)
|
||||
|
||||
|
||||
FUNCTIONS = {
|
||||
"news": news,
|
||||
"market_sentiment": market_sentiment,
|
||||
}
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: sentiment.py <function> [json_args]")
|
||||
sys.exit(1)
|
||||
|
||||
func_name = sys.argv[1]
|
||||
args = json.loads(sys.argv[2]) if len(sys.argv) > 2 else {}
|
||||
|
||||
if func_name not in FUNCTIONS:
|
||||
print(f"Unknown function: {func_name}")
|
||||
sys.exit(1)
|
||||
|
||||
result = FUNCTIONS[func_name](**args)
|
||||
print(result)
|
||||
@@ -0,0 +1,54 @@
|
||||
"""
|
||||
Single stock deep analysis module.
|
||||
Analyzes a stock across technical, fundamental, capital flow, and sentiment dimensions.
|
||||
|
||||
Usage: python3 stock_analyzer.py <function> <json_args>
|
||||
"""
|
||||
|
||||
import sys
|
||||
import json
|
||||
|
||||
|
||||
def analyze(symbol):
|
||||
"""Full analysis report. Stub - to be implemented."""
|
||||
return json.dumps({
|
||||
"status": "stub",
|
||||
"message": f"Analysis for {symbol} — not yet implemented",
|
||||
"symbol": symbol,
|
||||
"overall_score": None,
|
||||
"dimensions": {
|
||||
"technical": {"score": None, "details": None},
|
||||
"capital_flow": {"score": None, "details": None},
|
||||
"fundamental": {"score": None, "details": None},
|
||||
"sentiment": {"score": None, "details": None},
|
||||
},
|
||||
}, ensure_ascii=False)
|
||||
|
||||
|
||||
def technical(symbol):
|
||||
"""Technical analysis only. Stub - to be implemented."""
|
||||
return json.dumps({
|
||||
"status": "stub",
|
||||
"message": f"Technical analysis for {symbol} — not yet implemented",
|
||||
}, ensure_ascii=False)
|
||||
|
||||
|
||||
FUNCTIONS = {
|
||||
"analyze": analyze,
|
||||
"technical": technical,
|
||||
}
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: stock_analyzer.py <function> [json_args]")
|
||||
sys.exit(1)
|
||||
|
||||
func_name = sys.argv[1]
|
||||
args = json.loads(sys.argv[2]) if len(sys.argv) > 2 else {}
|
||||
|
||||
if func_name not in FUNCTIONS:
|
||||
print(f"Unknown function: {func_name}")
|
||||
sys.exit(1)
|
||||
|
||||
result = FUNCTIONS[func_name](**args)
|
||||
print(result)
|
||||
@@ -0,0 +1,68 @@
|
||||
"""
|
||||
Multi-factor A-share stock screening engine.
|
||||
|
||||
Usage: python3 stock_screener.py <function> <json_args>
|
||||
"""
|
||||
|
||||
import sys
|
||||
import json
|
||||
|
||||
|
||||
def multi_factor(strategy="comprehensive", sector=None, market_cap="all", top_n=10):
|
||||
"""Multi-factor scoring screen. Stub - to be implemented."""
|
||||
return json.dumps({
|
||||
"status": "stub",
|
||||
"message": f"Multi-factor screen (strategy={strategy}, sector={sector}) — not yet implemented",
|
||||
"strategy": strategy,
|
||||
"results": [],
|
||||
}, ensure_ascii=False)
|
||||
|
||||
|
||||
def strong(sector=None, top_n=20):
|
||||
"""Strong trend screen. Stub - to be implemented."""
|
||||
return json.dumps({
|
||||
"status": "stub",
|
||||
"message": "Strong trend screen — not yet implemented",
|
||||
"results": [],
|
||||
}, ensure_ascii=False)
|
||||
|
||||
|
||||
def breakout(lookback_days=60, top_n=20):
|
||||
"""Volume breakout screen. Stub - to be implemented."""
|
||||
return json.dumps({
|
||||
"status": "stub",
|
||||
"message": f"Breakout screen ({lookback_days}d) — not yet implemented",
|
||||
"results": [],
|
||||
}, ensure_ascii=False)
|
||||
|
||||
|
||||
def oversold(top_n=20):
|
||||
"""Oversold rebound screen. Stub - to be implemented."""
|
||||
return json.dumps({
|
||||
"status": "stub",
|
||||
"message": "Oversold screen — not yet implemented",
|
||||
"results": [],
|
||||
}, ensure_ascii=False)
|
||||
|
||||
|
||||
FUNCTIONS = {
|
||||
"multi_factor": multi_factor,
|
||||
"strong": strong,
|
||||
"breakout": breakout,
|
||||
"oversold": oversold,
|
||||
}
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) < 2:
|
||||
print("Usage: stock_screener.py <function> [json_args]")
|
||||
sys.exit(1)
|
||||
|
||||
func_name = sys.argv[1]
|
||||
args = json.loads(sys.argv[2]) if len(sys.argv) > 2 else {}
|
||||
|
||||
if func_name not in FUNCTIONS:
|
||||
print(f"Unknown function: {func_name}")
|
||||
sys.exit(1)
|
||||
|
||||
result = FUNCTIONS[func_name](**args)
|
||||
print(result)
|
||||
Reference in New Issue
Block a user